python+mysql.connector，demo實戰(zhàn)

說實話，其實沒有使用到框架，只是用了， python+mysql.connector模塊首先在開始虛擬環(huán)境：

(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

源代碼地址

代碼實現(xiàn) 創(chuàng)建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’, ’[email protected]’, ’1’);INSERT INTO `t_user` VALUES (’2’, ’scott’, ’3E6BC27A781F0AC08BCFD78CC3DCE4CA’, ’[email protected]’, ’1’);INSERT INTO `t_user` VALUES (’3’, ’test_1’, ’3E6BC27A781F0AC08BCFD78CC3DCE4CA’, ’[email protected]’, ’2’);INSERT INTO `t_user` VALUES (’4’, ’test_2’, ’3E6BC27A781F0AC08BCFD78CC3DCE4CA’, ’[email protected]’, ’2’);INSERT INTO `t_user` VALUES (’5’, ’test_3’, ’3E6BC27A781F0AC08BCFD78CC3DCE4CA’, ’[email protected]’, ’2’);INSERT INTO `t_user` VALUES (’6’, ’test_4’, ’3E6BC27A781F0AC08BCFD78CC3DCE4CA’, ’[email protected]’, ’2’);INSERT INTO `t_user` VALUES (’7’, ’test_5’, ’3E6BC27A781F0AC08BCFD78CC3DCE4CA’, ’[email protected]’, ’2’);INSERT INTO `t_user` VALUES (’8’, ’test_6’, ’3E6BC27A781F0AC08BCFD78CC3DCE4CA’, ’[email protected]’, ’2’);INSERT INTO `t_user` VALUES (’9’, ’test_7’, ’3E6BC27A781F0AC08BCFD78CC3DCE4CA’, ’[email protected]’, ’2’);INSERT INTO `t_user` VALUES (’10’, ’test_8’, ’3E6BC27A781F0AC08BCFD78CC3DCE4CA’, ’[email protected]’, ’2’);INSERT INTO `t_user` VALUES (’11’, ’test_9’, ’3E6BC27A781F0AC08BCFD78CC3DCE4CA’, ’[email protected]’, ’2’);INSERT INTO `t_user` VALUES (’12’, ’test_10’, ’3E6BC27A781F0AC08BCFD78CC3DCE4CA’, ’[email protected]’, ’2’);INSERT INTO `t_user` VALUES (’13’, ’test_11’, ’3E6BC27A781F0AC08BCFD78CC3DCE4CA’, ’[email protected]’, ’2’);

創(chuàng)建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()

創(chuàng)建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: # 創(chuàng)建私有對象 __user_dao = UserDao() # 創(chuàng)建登錄函數(shù) 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歡迎使用新聞管理系統(tǒng)') print(Fore.LIGHTBLUE_EX, 'nt=========================') print(Fore.LIGHTGREEN_EX, 'nt1.登錄系統(tǒng)') print(Fore.LIGHTGREEN_EX, 'nt2.退出系統(tǒng)') 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.退出系統(tǒng)') 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() # 查詢待審批新聞的總頁數(shù) 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() #查詢新聞總頁數(shù) 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 # 查詢待審批新聞的總頁數(shù) 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 # 查詢新聞總頁數(shù) 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歡迎使用新聞管理系統(tǒng)') print(Fore.LIGHTBLUE_EX, 'nt==================') print(Fore.LIGHTGREEN_EX,'nt1.登陸系統(tǒng)') print(Fore.LIGHTGREEN_EX,'nt2.退出系統(tǒng)') 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.退出系統(tǒng)') 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重復(fù)密碼:')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項目實戰(zhàn)的詳細內(nèi)容，更多關(guān)于python mysql項目實戰(zhàn)的資料請關(guān)注好吧啦網(wǎng)其它相關(guān)文章！