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

python+mysql.connector，demo实战

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

(vega-j-vI5SDr) (vega) D:testpython-mysqlpython-mysqlvega>pip install mysql.connectorProcessing c:usersadministratorappdatalocalpipcachewheels7b14395aad423666e827dfe9a1fbcd111ac17171e7c9865d570780cemysql_connector-2.2.9-cp39-cp39-win_amd64.whlInstalling collected packages: mysql.connectorSuccessfully 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 TransferSource Server : localhost_3306Source Server Version : 100120Source Host : localhost:3306Source Database : vegaTarget Server Type : MariaDBTarget Server Version : 100120File Encoding : 65001Date: 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 poolclass 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 Falseexcept 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 roleexcept 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 UserDaoclass 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:testpython-mysqlpython-mysqlvega>pip install coloramaCollecting colorama Using cached colorama-0.4.4-py2.py3-none-any.whl (16 kB)Installing collected packages: coloramaSuccessfully 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, Stylefrom getpass import getpassfrom service.user_service import UserServiceimport osimport sys__user_service = UserService()while True: os.system('cls') print(Fore.LIGHTBLUE_EX, 'nt=========================') print(Fore.LIGHTBLUE_EX, 'nt欢迎使用新闻管理系统') print(Fore.LIGHTBLUE_EX, 'nt=========================') print(Fore.LIGHTGREEN_EX, 'nt1.登录系统') print(Fore.LIGHTGREEN_EX, 'nt2.退出系统') print(Style.RESET_ALL) opt = input('nt输入操作编号：') if opt == '1':username = input('nt用户名:')password = getpass('nt密码：')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, 'nt1.新闻管理') print(Fore.LIGHTGREEN_EX, 'nt2.用户管理') print(Fore.LIGHTRED_EX, 'ntabck.退出登录') print(Fore.LIGHTRED_Ex, 'ntexit.退出系统') print(Style.RESET_ALL) opt = input('nt输入操作编号:')else: print('nt登录失败') elif opt == '2':sys.exit(0)

from db.mysql_db import poolclass 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 resultexcept 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_pageexcept 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 resultexcept 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_pageexcept 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 NewsDaoclass 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,initinit()from getpass import getpassfrom service.user_service import UserServicefrom service.news_service import NewsServicefrom service.role_service import RoleServiceimport osimport sysimport time__user_service=UserService()__news_service=NewsService()__role_service=RoleService()while True: os.system('cls') print(Fore.LIGHTBLUE_EX,'nt==================') print(Fore.LIGHTBLUE_EX,'nt欢迎使用新闻管理系统') print(Fore.LIGHTBLUE_EX, 'nt==================') print(Fore.LIGHTGREEN_EX,'nt1.登陆系统') print(Fore.LIGHTGREEN_EX,'nt2.退出系统') print(Style.RESET_ALL) opt=input('nt输入操作编号:') if opt=='1':username=input('nt用户名:')password=getpass('nt密码:')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,'nt1.新闻管理') print(Fore.LIGHTGREEN_EX, 'nt2.用户管理') print(Fore.LIGHTRED_EX, 'ntback.退出登陆') print(Fore.LIGHTRED_EX, 'ntexit.退出系统') print(Style.RESET_ALL) opt = input('nt输入操作编号:') if opt=='1':while True: os.system('cls') print(Fore.LIGHTGREEN_EX, 'nt1.审批新闻') print(Fore.LIGHTGREEN_EX, 'nt2.删除新闻') print(Fore.LIGHTRED_EX, 'ntback.返回上一层') print(Style.RESET_ALL) opt = input('nt输入操作编号:') if opt=='1':page=1while 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, 'nt%dt%st%st%s'%(index+1,one[1],one[2],one[3])) print(Fore.LIGHTBLUE_EX, 'nt-------------------') print(Fore.LIGHTBLUE_EX,'nt%d/%d'%(page,count_page)) print(Fore.LIGHTBLUE_EX, 'nt-------------------') print(Fore.LIGHTRED_EX, 'ntback.返回上一层') print(Fore.LIGHTRED_EX, 'ntprev.上一页') print(Fore.LIGHTRED_EX, 'ntnext.下一页') print(Style.RESET_ALL) opt = input('nt输入操作编号:') 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=1while 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, 'nt%dt%st%st%s'%(index+1,one[1],one[2],one[3])) print(Fore.LIGHTBLUE_EX, 'nt-------------------') print(Fore.LIGHTBLUE_EX,'nt%d/%d'%(page,count_page)) print(Fore.LIGHTBLUE_EX, 'nt-------------------') print(Fore.LIGHTRED_EX, 'ntback.返回上一层') print(Fore.LIGHTRED_EX, 'ntprev.上一页') print(Fore.LIGHTRED_EX, 'ntnext.下一页') print(Style.RESET_ALL) opt = input('nt输入操作编号:') 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, 'nt1.添加用户') print(Fore.LIGHTGREEN_EX, 'nt2.修改用户') print(Fore.LIGHTGREEN_EX, 'nt3.删除用户') print(Fore.LIGHTRED_EX, 'ntback.返回上一层') print(Style.RESET_ALL) opt = input('nt输入操作编号:') if opt=='back':break elif opt=='1':os.system('cls')username=input('nt用户名:')password = getpass('nt密码:')repassword=getpass('nt重复密码:')if password!=repassword: print('nt两次密码不一致(3秒自动返回)') time.sleep(3) continueemail=input('nt邮箱:')result=__role_service.search_list()for index in range(len(result)): one=result[index] print(Fore.LIGHTBLUE_EX,'nt%d.%s'%(index+1,one[1]))print(Style.RESET_ALL)opt=input('nt角色编号:')role_id=result[int(opt)-1][0]__user_service.insert(username,password,email,role_id)print('nt保存成功(3秒自动返回)')time.sleep(3) elif opt=='2':page = 1while 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, 'nt%dt%st%s' % (index + 1, one[1], one[2])) print(Fore.LIGHTBLUE_EX, 'nt-------------------') print(Fore.LIGHTBLUE_EX, 'nt%d/%d' % (page, count_page)) print(Fore.LIGHTBLUE_EX, 'nt-------------------') print(Fore.LIGHTRED_EX, 'ntback.返回上一层') print(Fore.LIGHTRED_EX, 'ntprev.上一页') print(Fore.LIGHTRED_EX, 'ntnext.下一页') print(Style.RESET_ALL) opt = input('nt输入操作编号:') 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('nt新用户名:')password = getpass('nt新密码:')repassword = getpass('nt再次输入密码:')if password!=repassword: print(Fore.LIGHTRED_EX,'nt两次密码不一致(3秒自动返回)') print(Style.RESET_ALL) time.sleep(3) breakemail = input('nt新邮箱:')result = __role_service.search_list()for index in range(len(result)): one = result[index] print(Fore.LIGHTBLUE_EX, 'nt%d.%s' % (index + 1, one[1]))print(Style.RESET_ALL)opt = input('nt角色编号:')role_id = result[int(opt) - 1][0]opt=input('nt是否保存(Y/N)')if opt=='Y' or opt=='y': __user_service.update(user_id,username,password,email,role_id) print('nt保存成功(3秒自动返回)') time.sleep(3) elif opt=='3':page = 1while 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, 'nt%dt%st%s' % (index + 1, one[1], one[2])) print(Fore.LIGHTBLUE_EX, 'nt-------------------') print(Fore.LIGHTBLUE_EX, 'nt%d/%d' % (page, count_page)) print(Fore.LIGHTBLUE_EX, 'nt-------------------') print(Fore.LIGHTRED_EX, 'ntback.返回上一层') print(Fore.LIGHTRED_EX, 'ntprev.上一页') print(Fore.LIGHTRED_EX, 'ntnext.下一页') print(Style.RESET_ALL) opt = input('nt输入操作编号:') 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('nt删除成功(3秒自动返回)')time.sleep(3) if opt==’back’:break; elif opt==’exit’:sys.exit(0)else: print('nt登录失败(3秒自动返回)') time.sleep(3) elif opt=='2':sys.exit(0)

以上就是python mysql项目实战的详细内容，更多关于python mysql项目实战的资料请关注好吧啦网其它相关文章！