python mysql项目实战及框架搭建过程

前言

python+mysql.connector，demo实战

框架搭建

说实话，其实没有使用到框架，只是用了， python+mysql.connector模块
首先在开始虚拟环境：

(vega-j-vI5SDr) (vega) D:\test\python-mysql\python-mysql\vega>pip install mysql.connector
Processing c:\users\administrator\appdata\local\pip\cache\wheels\7b\14\39\5aad423666e827dfe9a1fbcd111ac17171e7c9865d570780ce\mysql_connector-2.2.9-cp39-cp39-win_amd64.whl
Installing collected packages: mysql.connector
Successfully installed mysql.connector

源代码地址

代码实现 创建mysql连接池

#!/usr/bin/env python
# _*_ coding: utf-8 _*_
# @Time : 2021/6/6 13:16
# @Author : zhaocunwei
# @Version：V 0.1
# @File : mysql_db.py
# @desc :

import mysql.connector.pooling

__config = {
   "host": "localhost",
   "port": 3306,
   "user": "root",
   "password": "root",
   "database": "vega"
}

try:
   pool = mysql.connector.pooling.MySQLConnectionPool(
       **__config,
       pool_size=10
   )
except Exception as e:
   print(e)

SQL脚本：

/*
Navicat MariaDB Data Transfer

Source Server         : localhost_3306
Source Server Version : 100120
Source Host           : localhost:3306
Source Database       : vega

Target Server Type    : MariaDB
Target Server Version : 100120
File Encoding         : 65001

Date: 2018-11-27 19:35:26
*/

SET FOREIGN_KEY_CHECKS=0;

-- ----------------------------
-- Table structure for t_news
-- ----------------------------
DROP TABLE IF EXISTS `t_news`;
CREATE TABLE `t_news` (
 `id` int(10) unsigned NOT NULL AUTO_INCREMENT,
 `title` varchar(40) NOT NULL,
 `editor_id` int(10) unsigned NOT NULL,
 `type_id` int(10) unsigned NOT NULL,
 `content_id` char(12) NOT NULL,
 `is_top` tinyint(3) unsigned NOT NULL,
 `create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
 `update_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
 `state` enum('草稿','待审批','已审批','隐藏') NOT NULL,
 PRIMARY KEY (`id`),
 KEY `editor_id` (`editor_id`),
 KEY `type_id` (`type_id`),
 KEY `state` (`state`),
 KEY `create_time` (`create_time`),
 KEY `is_top` (`is_top`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;

-- ----------------------------
-- Records of t_news
-- ----------------------------
INSERT INTO `t_news` VALUES ('1', '新闻标题1', '2', '1', '1', '1', '2018-11-22 18:55:56', '2018-11-22 18:55:56', '待审批');

-- ----------------------------
-- Table structure for t_role
-- ----------------------------
DROP TABLE IF EXISTS `t_role`;
CREATE TABLE `t_role` (
 `id` int(10) unsigned NOT NULL AUTO_INCREMENT,
 `role` varchar(20) NOT NULL,
 PRIMARY KEY (`id`),
 UNIQUE KEY `role` (`role`)
) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8;

-- ----------------------------
-- Records of t_role
-- ----------------------------
INSERT INTO `t_role` VALUES ('2', '新闻编辑');
INSERT INTO `t_role` VALUES ('1', '管理员');

-- ----------------------------
-- Table structure for t_type
-- ----------------------------
DROP TABLE IF EXISTS `t_type`;
CREATE TABLE `t_type` (
 `id` int(10) unsigned NOT NULL AUTO_INCREMENT,
 `type` varchar(20) NOT NULL,
 PRIMARY KEY (`id`),
 UNIQUE KEY `type` (`type`)
) ENGINE=InnoDB AUTO_INCREMENT=6 DEFAULT CHARSET=utf8;

-- ----------------------------
-- Records of t_type
-- ----------------------------
INSERT INTO `t_type` VALUES ('2', '体育');
INSERT INTO `t_type` VALUES ('5', '历史');
INSERT INTO `t_type` VALUES ('4', '娱乐');
INSERT INTO `t_type` VALUES ('3', '科技');
INSERT INTO `t_type` VALUES ('1', '要闻');

-- ----------------------------
-- Table structure for t_user
-- ----------------------------
DROP TABLE IF EXISTS `t_user`;
CREATE TABLE `t_user` (
 `id` int(10) unsigned NOT NULL AUTO_INCREMENT,
 `username` varchar(20) NOT NULL,
 `password` varchar(500) NOT NULL,
 `email` varchar(100) NOT NULL,
 `role_id` int(10) unsigned NOT NULL,
 PRIMARY KEY (`id`),
 UNIQUE KEY `username` (`username`),
 KEY `username_2` (`username`)
) ENGINE=InnoDB AUTO_INCREMENT=14 DEFAULT CHARSET=utf8;

-- ----------------------------
-- Records of t_user
-- ----------------------------
INSERT INTO `t_user` VALUES ('1', 'admin', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'admin@163.com', '1');
INSERT INTO `t_user` VALUES ('2', 'scott', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'scott@163.com', '1');
INSERT INTO `t_user` VALUES ('3', 'test_1', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'test_1@163.com', '2');
INSERT INTO `t_user` VALUES ('4', 'test_2', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'test_2@163.com', '2');
INSERT INTO `t_user` VALUES ('5', 'test_3', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'test_3@163.com', '2');
INSERT INTO `t_user` VALUES ('6', 'test_4', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'test_4@163.com', '2');
INSERT INTO `t_user` VALUES ('7', 'test_5', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'test_5@163.com', '2');
INSERT INTO `t_user` VALUES ('8', 'test_6', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'test_6@163.com', '2');
INSERT INTO `t_user` VALUES ('9', 'test_7', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'test_7@163.com', '2');
INSERT INTO `t_user` VALUES ('10', 'test_8', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'test_8@163.com', '2');
INSERT INTO `t_user` VALUES ('11', 'test_9', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'test_9@163.com', '2');
INSERT INTO `t_user` VALUES ('12', 'test_10', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'test_10@163.com', '2');
INSERT INTO `t_user` VALUES ('13', 'test_11', '3E6BC27A781F0AC08BCFD78CC3DCE4CA', 'test_11@163.com', '2');

创建DAO程序

#!/usr/bin/env python
# _*_ coding: utf-8 _*_
# @Time : 2021/6/6 13:24
# @Author : zhaocunwei
# @Version：V 0.1
# @File : user_dao.py
# @desc : 用户

from db.mysql_db import pool

class UserDao:
   # 验证用户登录
   def login(self, username, password):
       try:
           con = pool.get_connection()
           cursor = con.cursor()
           sql = "SELECT COUNT(*) FROM t_user WHERE username=％s AND " \
                 "AES_DECRYPT(UNHEX(password),'HelloWorld')=％s"
           cursor.execute(sql, (username, password))
           count = cursor.fetchone()[0]
           return True if count == 1 else False
       except Exception as e:
           print(e)
       finally:
           if "con" in dir():
               con.close()

# 查询用户角色
   def search_user_role(self, username):
       try:
           con = pool.get_connection()
           cursor = con.cursor()
           sql = "SELECT r.role FROM t_user u JOIN t_role r ON u.role_id=r.id" \
                 "WHERE u.username=％s"
           cursor.execute(sql, (username))
           role = cursor.fetchone()[0]
           return role
       except Exception as e:
           print(e)
       finally:
           if "con" in dir():
               con.close()

创建service层程序

#!/usr/bin/env python
# _*_ coding: utf-8 _*_
# @Time : 2021/6/6 13:57
# @Author : zhaocunwei
# @Version：V 0.1
# @File : user_service.py
# @desc :

from db.user_dao import UserDao

class UserService:
   # 创建私有对象
   __user_dao = UserDao()

# 创建登录函数
   def login(self, username, password):
       result = self.__user_dao.login(username, password)
       return result

# 查询用户角色
   def search_user_role(self, username):
       role = self.__user_dao.search_user_role(username)
       return role

安装变色的模块，O(∩_∩)O哈哈~

(vega-j-vI5SDr) (vega) D:\test\python-mysql\python-mysql\vega>pip install colorama
Collecting colorama
 Using cached colorama-0.4.4-py2.py3-none-any.whl (16 kB)
Installing collected packages: colorama
Successfully installed colorama-0.4.4

CMD模拟登陆

#!/usr/bin/env python
# _*_ coding: utf-8 _*_
# @Time : 2021/6/6 14:08
# @Author : zhaocunwei
# @Version：V 0.1
# @File : app.py
# @desc : 控制台程序

from colorama import Fore, Style
from getpass import getpass
from service.user_service import UserService
import os
import sys

__user_service = UserService()

while True:
   os.system("cls")
   print(Fore.LIGHTBLUE_EX, "\n\t=========================")
   print(Fore.LIGHTBLUE_EX, "\n\t欢迎使用新闻管理系统")
   print(Fore.LIGHTBLUE_EX, "\n\t=========================")
   print(Fore.LIGHTGREEN_EX, "\n\t1.登录系统")
   print(Fore.LIGHTGREEN_EX, "\n\t2.退出系统")
   print(Style.RESET_ALL)
   opt = input("\n\t输入操作编号：")
   if opt == "1":
       username = input("\n\t用户名:")
       password = getpass("\n\t密码：")
       result = __user_service.login(username, password)
       # 登录成功
       if result == True:
           # 查询角色
           role = __user_service.search_user_role(username)
           os.system("cls")
           while True:
               if role == "新闻编辑":
                   print("test")
               elif role == "管理员":
                   print(Fore.LIGHTGREEN_EX, "\n\t1.新闻管理")
                   print(Fore.LIGHTGREEN_EX, "\n\t2.用户管理")
                   print(Fore.LIGHTRED_EX, "\n\tabck.退出登录")
                   print(Fore.LIGHTRED_Ex, "\n\texit.退出系统")
                   print(Style.RESET_ALL)
                   opt = input("\n\t输入操作编号:")
       else:
           print("\n\t登录失败")
   elif opt == "2":
       sys.exit(0)

from db.mysql_db import pool

class NewsDao:
   #查询待审批新闻列表
   def search_unreview_list(self,page):
       try:
           con=pool.get_connection()
           cursor=con.cursor()
           sql="SELECT n.id,n.title,t.type,u.username " \
               "FROM t_news n JOIN t_type t ON n.type_id=t.id " \
               "JOIN t_user u ON n.editor_id=u.id " \
               "WHERE n.state=％s " \
               "ORDER BY n.create_time DESC " \
               "LIMIT ％s,％s"
           cursor.execute(sql,("待审批",(page-1)*10,10))
           result=cursor.fetchall()
           return result
       except Exception as e:
           print(e)
       finally:
           if "con" in dir():
               con.close()

# 查询待审批新闻的总页数
   def search_unreview_count_page(self):
       try:
           con=pool.get_connection()
           cursor=con.cursor()
           sql="SELECT CEIL(COUNT(*)/10) FROM t_news WHERE state=％s"
           cursor.execute(sql,["待审批"])
           count_page=cursor.fetchone()[0]
           return count_page
       except Exception as e:
           print(e)
       finally:
           if "con" in dir():
               con.close()
   #审批新闻
   def update_unreview_news(self,id):
       try:
           con = pool.get_connection()
           con.start_transaction()
           cursor=con.cursor()
           sql="UPDATE t_news SET state=％s WHERE id=％s"
           cursor.execute(sql,("已审批",id))
           con.commit()
       except Exception as e:
           if "con" in dir():
               con.rollback()
           print(e)
       finally:
           if "con" in dir():
               con.close()

#查询新闻列表
   def search_list(self,page):
       try:
           con=pool.get_connection()
           cursor=con.cursor()
           sql="SELECT n.id,n.title,t.type,u.username " \
               "FROM t_news n JOIN t_type t ON n.type_id=t.id " \
               "JOIN t_user u ON n.editor_id=u.id " \
               "ORDER BY n.create_time DESC " \
               "LIMIT ％s,％s"
           cursor.execute(sql,((page-1)*10,10))
           result=cursor.fetchall()
           return result
       except Exception as e:
           print(e)
       finally:
           if "con" in dir():
               con.close()

#查询新闻总页数
   def search_count_page(self):
       try:
           con=pool.get_connection()
           cursor=con.cursor()
           sql="SELECT CEIL(COUNT(*)/10) FROM t_news"
           cursor.execute(sql)
           count_page=cursor.fetchone()[0]
           return count_page
       except Exception as e:
           print(e)
       finally:
           if "con" in dir():
               con.close()

#删除新闻
   def delete_by_id(self,id):
       try:
           con = pool.get_connection()
           con.start_transaction()
           cursor=con.cursor()
           sql="DELETE FROM t_news WHERE id=％s"
           cursor.execute(sql,[id])
           con.commit()
       except Exception as e:
           if "con" in dir():
               con.rollback()
           print(e)
       finally:
           if "con" in dir():
               con.close()

from db.news_dao import NewsDao

class NewsService:
   __news_dao=NewsDao()

# 查询待审批新闻列表
   def search_unreview_list(self,page):
       result=self.__news_dao.search_unreview_list(page)
       return result

# 查询待审批新闻的总页数
   def search_unreview_count_page(self):
       count_page=self.__news_dao.search_unreview_count_page()
       return count_page

# 审批新闻
   def update_unreview_news(self, id):
       self.__news_dao.update_unreview_news(id)

#查询新闻列表
   def search_list(self, page):
       result=self.__news_dao.search_list(page)
       return result

# 查询新闻总页数
   def search_count_page(self):
       count_page=self.__news_dao.search_count_page()
       return count_page

# 删除新闻
   def delete_by_id(self, id):
       self.__news_dao.delete_by_id(id)

from colorama import Fore,Style,init
init()
from getpass import getpass
from service.user_service import UserService
from service.news_service import NewsService
from service.role_service import RoleService
import os
import sys
import time

__user_service=UserService()
__news_service=NewsService()
__role_service=RoleService()

while True:
   os.system("cls")
   print(Fore.LIGHTBLUE_EX,"\n\t==================")
   print(Fore.LIGHTBLUE_EX,"\n\t欢迎使用新闻管理系统")
   print(Fore.LIGHTBLUE_EX, "\n\t==================")
   print(Fore.LIGHTGREEN_EX,"\n\t1.登陆系统")
   print(Fore.LIGHTGREEN_EX,"\n\t2.退出系统")
   print(Style.RESET_ALL)
   opt=input("\n\t输入操作编号:")
   if opt=="1":
       username=input("\n\t用户名:")
       password=getpass("\n\t密码:")
       result=__user_service.login(username,password)
       #登陆成功
       if result==True:
           #查询角色
           role=__user_service.search_user_role(username)
           while True:
               os.system("cls")
               if role=="新闻编辑":
                   print('test')
               elif role=="管理员":
                   print(Fore.LIGHTGREEN_EX,"\n\t1.新闻管理")
                   print(Fore.LIGHTGREEN_EX, "\n\t2.用户管理")
                   print(Fore.LIGHTRED_EX, "\n\tback.退出登陆")
                   print(Fore.LIGHTRED_EX, "\n\texit.退出系统")
                   print(Style.RESET_ALL)
                   opt = input("\n\t输入操作编号:")
                   if opt=="1":
                       while True:
                           os.system("cls")
                           print(Fore.LIGHTGREEN_EX, "\n\t1.审批新闻")
                           print(Fore.LIGHTGREEN_EX, "\n\t2.删除新闻")
                           print(Fore.LIGHTRED_EX, "\n\tback.返回上一层")
                           print(Style.RESET_ALL)
                           opt = input("\n\t输入操作编号:")
                           if opt=="1":
                               page=1
                               while True:
                                   os.system("cls")
                                   count_page=__news_service.search_unreview_count_page()
                                   result=__news_service.search_unreview_list(page)
                                   for index in range(len(result)):
                                       one=result[index]
                                       print(Fore.LIGHTBLUE_EX, "\n\t％d\t％s\t％s\t％s"％(index+1,one[1],one[2],one[3]))
                                   print(Fore.LIGHTBLUE_EX, "\n\t-------------------")
                                   print(Fore.LIGHTBLUE_EX,"\n\t％d/％d"％(page,count_page))
                                   print(Fore.LIGHTBLUE_EX, "\n\t-------------------")
                                   print(Fore.LIGHTRED_EX, "\n\tback.返回上一层")
                                   print(Fore.LIGHTRED_EX, "\n\tprev.上一页")
                                   print(Fore.LIGHTRED_EX, "\n\tnext.下一页")
                                   print(Style.RESET_ALL)
                                   opt = input("\n\t输入操作编号:")
                                   if opt=="back":
                                       break
                                   elif opt=="prev" and page>1:
                                       page-=1
                                   elif opt=="next" and page<count_page:
                                       page+=1
                                   elif int(opt)>=1 and int(opt)<=10:
                                       news_id=result[int(opt)-1][0]
                                       __news_service.update_unreview_news(news_id)
                           elif opt=="2":
                               page=1
                               while True:
                                   os.system("cls")
                                   count_page=__news_service.search_count_page()
                                   result=__news_service.search_list(page)
                                   for index in range(len(result)):
                                       one=result[index]
                                       print(Fore.LIGHTBLUE_EX, "\n\t％d\t％s\t％s\t％s"％(index+1,one[1],one[2],one[3]))
                                   print(Fore.LIGHTBLUE_EX, "\n\t-------------------")
                                   print(Fore.LIGHTBLUE_EX,"\n\t％d/％d"％(page,count_page))
                                   print(Fore.LIGHTBLUE_EX, "\n\t-------------------")
                                   print(Fore.LIGHTRED_EX, "\n\tback.返回上一层")
                                   print(Fore.LIGHTRED_EX, "\n\tprev.上一页")
                                   print(Fore.LIGHTRED_EX, "\n\tnext.下一页")
                                   print(Style.RESET_ALL)
                                   opt = input("\n\t输入操作编号:")
                                   if opt=="back":
                                       break
                                   elif opt=="prev" and page>1:
                                       page-=1
                                   elif opt=="next" and page<count_page:
                                       page+=1
                                   elif int(opt)>=1 and int(opt)<=10:
                                       news_id=result[int(opt)-1][0]
                                       __news_service.delete_by_id(news_id)
                           elif opt=="back":
                               break
                   elif opt=="2":
                       while True:
                           os.system("cls")
                           print(Fore.LIGHTGREEN_EX, "\n\t1.添加用户")
                           print(Fore.LIGHTGREEN_EX, "\n\t2.修改用户")
                           print(Fore.LIGHTGREEN_EX, "\n\t3.删除用户")
                           print(Fore.LIGHTRED_EX, "\n\tback.返回上一层")
                           print(Style.RESET_ALL)
                           opt = input("\n\t输入操作编号:")
                           if opt=="back":
                               break
                           elif opt=="1":
                               os.system("cls")
                               username=input("\n\t用户名:")
                               password = getpass("\n\t密码:")
                               repassword=getpass("\n\t重复密码:")
                               if password!=repassword:
                                   print("\n\t两次密码不一致(3秒自动返回)")
                                   time.sleep(3)
                                   continue
                               email=input("\n\t邮箱:")
                               result=__role_service.search_list()
                               for index in range(len(result)):
                                   one=result[index]
                                   print(Fore.LIGHTBLUE_EX,"\n\t％d.％s"％(index+1,one[1]))
                               print(Style.RESET_ALL)
                               opt=input("\n\t角色编号:")
                               role_id=result[int(opt)-1][0]
                               __user_service.insert(username,password,email,role_id)
                               print("\n\t保存成功(3秒自动返回)")
                               time.sleep(3)
                           elif opt=="2":
                               page = 1
                               while True:
                                   os.system("cls")
                                   count_page = __user_service.search_count_page()
                                   result = __user_service.search_list(page)
                                   for index in range(len(result)):
                                       one = result[index]
                                       print(Fore.LIGHTBLUE_EX,
                                             "\n\t％d\t％s\t％s" ％ (index + 1, one[1], one[2]))
                                   print(Fore.LIGHTBLUE_EX, "\n\t-------------------")
                                   print(Fore.LIGHTBLUE_EX, "\n\t％d/％d" ％ (page, count_page))
                                   print(Fore.LIGHTBLUE_EX, "\n\t-------------------")
                                   print(Fore.LIGHTRED_EX, "\n\tback.返回上一层")
                                   print(Fore.LIGHTRED_EX, "\n\tprev.上一页")
                                   print(Fore.LIGHTRED_EX, "\n\tnext.下一页")
                                   print(Style.RESET_ALL)
                                   opt = input("\n\t输入操作编号:")
                                   if opt == "back":
                                       break
                                   elif opt == "prev" and page > 1:
                                       page -= 1
                                   elif opt == "next" and page < count_page:
                                       page += 1
                                   elif int(opt) >= 1 and int(opt) <= 10:
                                       os.system("cls")
                                       user_id=result[int(opt)-1][0]
                                       username = input("\n\t新用户名:")
                                       password = getpass("\n\t新密码:")
                                       repassword = getpass("\n\t再次输入密码:")
                                       if password!=repassword:
                                           print(Fore.LIGHTRED_EX,"\n\t两次密码不一致(3秒自动返回)")
                                           print(Style.RESET_ALL)
                                           time.sleep(3)
                                           break
                                       email = input("\n\t新邮箱:")
                                       result = __role_service.search_list()
                                       for index in range(len(result)):
                                           one = result[index]
                                           print(Fore.LIGHTBLUE_EX, "\n\t％d.％s" ％ (index + 1, one[1]))
                                       print(Style.RESET_ALL)
                                       opt = input("\n\t角色编号:")
                                       role_id = result[int(opt) - 1][0]
                                       opt=input("\n\t是否保存(Y/N)")
                                       if opt=="Y" or opt=="y":
                                           __user_service.update(user_id,username,password,email,role_id)
                                           print("\n\t保存成功(3秒自动返回)")
                                           time.sleep(3)
                           elif opt=="3":
                               page = 1
                               while True:
                                   os.system("cls")
                                   count_page = __user_service.search_count_page()
                                   result = __user_service.search_list(page)
                                   for index in range(len(result)):
                                       one = result[index]
                                       print(Fore.LIGHTBLUE_EX,
                                             "\n\t％d\t％s\t％s" ％ (index + 1, one[1], one[2]))
                                   print(Fore.LIGHTBLUE_EX, "\n\t-------------------")
                                   print(Fore.LIGHTBLUE_EX, "\n\t％d/％d" ％ (page, count_page))
                                   print(Fore.LIGHTBLUE_EX, "\n\t-------------------")
                                   print(Fore.LIGHTRED_EX, "\n\tback.返回上一层")
                                   print(Fore.LIGHTRED_EX, "\n\tprev.上一页")
                                   print(Fore.LIGHTRED_EX, "\n\tnext.下一页")
                                   print(Style.RESET_ALL)
                                   opt = input("\n\t输入操作编号:")
                                   if opt == "back":
                                       break
                                   elif opt == "prev" and page > 1:
                                       page -= 1
                                   elif opt == "next" and page < count_page:
                                       page += 1
                                   elif int(opt) >= 1 and int(opt) <= 10:
                                       os.system("cls")
                                       user_id=result[int(opt)-1][0]
                                       __user_service.delete_by_id(user_id)
                                       print("\n\t删除成功(3秒自动返回)")
                                       time.sleep(3)

if opt=='back':
                       break;
                   elif opt=='exit':
                       sys.exit(0)

else:
           print("\n\t登录失败(3秒自动返回)")
           time.sleep(3)

elif opt=="2":
       sys.exit(0)

来源：https://blog.csdn.net/qq_32370913/article/details/117625741