前言
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)
以上就是python mysql项目实战的详细内容,更多关于python mysql项目实战的资料请关注自学编程网其它相关文章!