自学教程：python mysql项目实战及框架搭建过程

(vega-j-vI5SDr) (vega) D:/test/python-mysql/python-mysql/vega>pip install mysql.connectorProcessing c:/users/administrator/appdata/local/pip/cache/wheels/7b/14/39/5aad423666e827dfe9a1fbcd111ac17171e7c9865d570780ce/mysql_connector-2.2.9-cp39-cp39-win_amd64.whlInstalling collected packages: mysql.connectorSuccessfully 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 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');

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

(vega-j-vI5SDr) (vega) D:/test/python-mysql/python-mysql/vega>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

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