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

 (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

 #!/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)

 /* 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');

 #!/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()

 #!/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

 (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

 #!/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=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=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 = 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 = 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)