python mysql項(xiàng)目實(shí)戰(zhàn)及框架搭建過程
python+mysql.connector,demo實(shí)戰(zhàn)
框架搭建
說實(shí)話,其實(shí)沒有使用到框架,只是用了, python+mysql.connector模塊
首先在開始虛擬環(huán)境:
(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
源代碼地址
代碼實(shí)現(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 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', '新聞標(biāo)題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');
創(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 pool class UserDao: # 驗(yàn)證用戶登錄 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()
創(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 UserDao class UserService: # 創(chuàng)建私有對(duì)象 __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:\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 : 控制臺(tái)程序 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歡迎使用新聞管理系統(tǒng)") print(Fore.LIGHTBLUE_EX, "\n\t=========================") print(Fore.LIGHTGREEN_EX, "\n\t1.登錄系統(tǒng)") print(Fore.LIGHTGREEN_EX, "\n\t2.退出系統(tǒng)") print(Style.RESET_ALL) opt = input("\n\t輸入操作編號(hào):") 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.退出系統(tǒng)") print(Style.RESET_ALL) opt = input("\n\t輸入操作編號(hào):") 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() # 查詢待審批新聞的總頁數(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_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() #查詢新聞總頁數(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_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 # 查詢待審批新聞的總頁數(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,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歡迎使用新聞管理系統(tǒng)") print(Fore.LIGHTBLUE_EX, "\n\t==================") print(Fore.LIGHTGREEN_EX,"\n\t1.登陸系統(tǒng)") print(Fore.LIGHTGREEN_EX,"\n\t2.退出系統(tǒng)") print(Style.RESET_ALL) opt=input("\n\t輸入操作編號(hào):") 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.退出系統(tǒng)") print(Style.RESET_ALL) opt = input("\n\t輸入操作編號(hào):") 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輸入操作編號(hào):") 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輸入操作編號(hào):") 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輸入操作編號(hào):") 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輸入操作編號(hào):") if opt=="back": break elif opt=="1": os.system("cls") username=input("\n\t用戶名:") password = getpass("\n\t密碼:") repassword=getpass("\n\t重復(fù)密碼:") if password!=repassword: print("\n\t兩次密碼不一致(3秒自動(dòng)返回)") 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角色編號(hào):") role_id=result[int(opt)-1][0] __user_service.insert(username,password,email,role_id) print("\n\t保存成功(3秒自動(dòng)返回)") 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輸入操作編號(hào):") 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秒自動(dòng)返回)") 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角色編號(hào):") 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秒自動(dòng)返回)") 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輸入操作編號(hào):") 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秒自動(dòng)返回)") time.sleep(3) if opt=='back':break; elif opt=='exit':sys.exit(0) else: print("\n\t登錄失敗(3秒自動(dòng)返回)") time.sleep(3) elif opt=="2": sys.exit(0)
以上就是python mysql項(xiàng)目實(shí)戰(zhàn)的詳細(xì)內(nèi)容,更多關(guān)于python mysql項(xiàng)目實(shí)戰(zhàn)的資料請(qǐng)關(guān)注本站其它相關(guān)文章!
版權(quán)聲明:本站文章來源標(biāo)注為YINGSOO的內(nèi)容版權(quán)均為本站所有,歡迎引用、轉(zhuǎn)載,請(qǐng)保持原文完整并注明來源及原文鏈接。禁止復(fù)制或仿造本網(wǎng)站,禁止在非www.sddonglingsh.com所屬的服務(wù)器上建立鏡像,否則將依法追究法律責(zé)任。本站部分內(nèi)容來源于網(wǎng)友推薦、互聯(lián)網(wǎng)收集整理而來,僅供學(xué)習(xí)參考,不代表本站立場(chǎng),如有內(nèi)容涉嫌侵權(quán),請(qǐng)聯(lián)系alex-e#qq.com處理。