快速入手-搭建Flask框架封装mysql并结合业务实际情况使用

1、安装包

pip install pymysql

pip install dbutils

3、在flask_project文件夹里创建__init__.py

from flask import Flask, redirect, request, session
def auth():# 拦截器print("拦截器")# js、css、img不拦截处理if request.path.startswith("/static"):# 继续往后执行，不拦截returnif request.path == "/login":# 继续往后执行，不拦截returnuser_inf = session.get("user_inf")if user_inf:returnreturn redirect("/login")def create_app():app = Flask(__name__)# session加密app.secret_key = "asdfae233daer2rewdfa"from .views import demofrom .views import loginapp.register_blueprint(demo.do)app.register_blueprint(login.lg)app.before_request(auth)return app

4、在flask_project同级目录创建app.py

from flask_project import create_appapp = create_app()if __name__ == "__main__":app.run(host="127.0.0.1", port=5000, debug=True)

5、在utils里创建mysql的工具db.py

import pymysql
from dbutils.pooled_db import PooledDB
from pymysql import cursorsPOOL = PooledDB(creator=pymysql,  # 使用链接数据库的模块maxconnections=20,  # 连接池允许的最大连接数，0和None表示不限制连接数mincached=2,  # 初始化时，链接池中至少创建的空闲的链接，0表示不创建maxcached=5,  # 链接池中空闲链接的最大数量blocking=True,  # 链接池中如果没有可用连接后，是否阻塞等待setsession=[],  # 设置会话,开始回话前执行的命令列表。ping=0,host="127.0.0.1",port=3306,user="root",password="1234567890",database="qyd_dzfp",charset="utf8",
)def fetch_one(sql, params):conn = POOL.connection()cursor = conn.cursor(cursor=cursors.DictCursor)cursor.execute(sql, params)result = cursor.fetchone()# 不是关闭数据库链接，将该链接交还给连接池cursor.close()conn.close()return result

6、在views里创建login.py

from flask import Blueprint, session
from utils import db# 蓝图对象
lg = Blueprint("login", __name__)@lg.route("/login", methods=["GET"])
def login():user_id = "1837035517405904898"status = "1"user_dict = db.fetch_one("select * from sys_user where user_id = %s and status = %s",[user_id, status],)# print(user_dict)name = user_dict["realname"]if user_dict:session["user_inf"] = {"status": user_dict["status"],"realname": user_dict["realname"],}return "登录成功"else:return "用户不存在"

7、在views里创建demo.py

from flask import Blueprint, session, jsonify# 蓝图对象
do = Blueprint("demo", __name__)@do.route("/demo/getuser", methods=["GET"])
def getuser():user_inf = session.get("user_inf")if not user_inf:return jsonify({"status": 0, "msg": "用户未登录"})return user_inf["realname"]

8、运行项目

首次访问127.0.01:5000/demo/getuser,会跳转到登录界面，模拟自动登录成功