当前位置：首页 > news >正文

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

news 来源：原创 2025/6/15 13:46:25

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"):
        # 继续往后执行，不拦截
        return
    if request.path == "/login":
        # 继续往后执行，不拦截
        return
    user_inf = session.get("user_inf")
    if user_inf:
        return
    return redirect("/login")


def create_app():
    app = Flask(__name__)
    # session加密
    app.secret_key = "asdfae233daer2rewdfa"

    from .views import demo
    from .views import login

    app.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_app

app = 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 cursors

POOL = 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,会跳转到登录界面，模拟自动登录成功