【python| 框架】flask速通- 惊觉

一.介绍

Flask 是一款基于 Python 开发的轻量级 Web 应用框架，核心依赖 Werkzeug（WSGI 工具集，负责路由、请求响应处理）和 Jinja2（模板引擎，负责页面渲染），属于典型的「微框架（Micro-Framework）」。

它的设计理念是「核心精简、按需扩展」：框架本身只提供 Web 开发的基础能力，复杂功能通过第三方扩展实现，开发者可以自由选择技术栈，灵活度极高。

安装：

pip install flask

简单示例：

from flask import Flask

app = Flask(__name__)

@app.route('/demo')
def hello():
    return 'Hello, Flask!'

if __name__ == '__main__':
    app.run(debug=True)

程序运行后得到网址。

访问网址时

这是因为仅仅访问了网址，若要显示具体内容则需要在网址上加/demo

本篇将从无耗时版，耗时版来解析。

二.无耗时版

无耗时版flask流程图

2.1获取请求数据

2.1.1 获取 URL 查询参数（Query String）

 age = request.args.get('age')
 pwd = request.args.get('pwd')
 print(age, pwd)

2.1.2 获取表单数据（Form Data）

xx = request.form.get('xx')
yy = request.form.get('yy')
print(xx,yy)

2.1.3 JSON 格式

2.1.3.1 获取 JSON 格式的请求数据

print(request.json)

request.json 是一个属性，它会自动解析请求体中的 JSON 数据，并返回 Python 字典（或列表等）。

要求请求的 Content-Type 必须是 application/json，否则会返回 None。

2.1.3.2 返回 JSON 响应

 return jsonify({'status':'Ture','data':'sucessful'})

jsonify() 是 Flask 提供的一个函数，它会把传入的字典转换成 JSON 字符串，并自动设置响应头的 Content-Type 为 application/json。

浏览器浏览器里只能直接发起 GET 请求，默认是模拟不出来post请求，所以需要用到postman

postman下载官网https://www.postman.com/downloads/

根据需求自己进行测试 URL 参数，测试表单提交，测试 JSON 数据等一系列操作。

2.2 直接访问api并返回数据

接收一个字符串 + 拼接密钥 + MD5 签名并立即返回签名的完整接口逻辑。

import hashlib
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route("/bili", methods=["POST"])
def bili():
    ordered_string = request.json.get("ordered_string")
    if not ordered_string:
        return jsonify({"status": False, "error": "参数错误"}) 

    # 拼接密钥，生成MD5签名
    encrypt_string = ordered_string + "560c52ccd288fed045859ed18bffd973"
    obj = hashlib.md5(encrypt_string.encode('utf-8'))
    sign = obj.hexdigest()

    return jsonify({"status": True, "data": sign})

if __name__ == '__main__':
    app.run(host="127.0.0.1", port=5000)

对于该代码来说，postman测试时

请求方法：POST

URL：http://127.0.0.1:5000/bili

Body 选择 raw → JSON

发送

可得到回复

2.3 文件版本授权

2.3.1 读取白名单

DB_FILE = "db.txt"

def get_user_dict():
    info_dict = {}
    try:
        with open(DB_FILE, 'r', encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                parts = line.split('\t')          
                if len(parts) >= 2:
                    token = parts[0]
                    username = parts[1]
                    info_dict[token] = username
    except FileNotFoundError:
        pass   
    return info_dict

2.3.2 在接口中校验 token

@app.route("/bo", methods=["POST"])
def bo():
    token = request.args.get("token")
    if not token:
        return jsonify({"status": False, "error": "认证失败：缺少token参数"}), 401

    # 加载白名单
    user_dict = get_user_dict()
    if token not in user_dict:
        return jsonify({"status": False, "error": "认证失败：token无效"}), 401

    # ... 后续业务逻辑（如签名计算）

文件授权（db.txt）的作用详解
核心用途：接口身份鉴权（登录校验）
这个 db.txt 是白名单凭证库，用来控制谁能调用你的 MD5 加密接口：
只有文件里登记过的 token，才能成功请求接口；
没登记的 token、乱填的 token，直接返回 401 认证失败，拒绝访问；
每个 token 绑定一个用户名，方便区分是谁在调用接口。

2.4 数据库版本

2.4.1 跨域

from flask_cors import CORS
CORS(app)

前端页面（如 http://localhost:8080）调用后端 http://127.0.0.1:5000，浏览器同源策略拦截请求，报跨域错误。
flask-cors 专门解决该问题。

2.4.2 数据库配置

DB_CONFIG = {
    "host": "127.0.0.1",
    "port": 3306,
    "user": "root",
    "password": "密码",
    "database": "数据库名",
    "charset": "utf8mb4"
}

2.4.3 数据库查询

def fetch_one(sql, params=None):
    conn = None
    cursor = None
    try:
        conn = pymysql.connect(**DB_CONFIG)
        cursor = conn.cursor()
        cursor.execute(sql, params or ())
        return cursor.fetchone()
    except MySQLError as e:
        app.logger.error(f"数据库异常: {str(e)}")
        return None
    finally:
        if cursor:
            cursor.close()
        if conn:
            conn.close()

每次调用都创建新连接，执行完立即关闭。

2.4.4 接口核心逻辑

@app.route("/boo", methods=["POST"])
def boo():
    try:
        token = request.args.get("token")
        if not token:
            return jsonify({"status": False, "error": "认证失败：缺少token参数"}), 401

        # 数据库校验 token，同时取出用户名
        user_row = fetch_one("SELECT token, name FROM flask WHERE token=%s", (token,))
        if not user_row:
            return jsonify({"status": False, "error": "认证失败：token无效"}), 401
        username = user_row[1]

        # JSON 格式校验、参数校验 ...
        # MD5 签名计算 ...
        # 返回结果
    except Exception as e:
        return jsonify({"status": False, "error": "服务器内部错误"}), 500

从 URL 参数获取 token

数据库白名单验证：SELECT token, name FROM flask WHERE token=%s

获取用户名后附在响应中，方便识别调用者

2.5 数据库连接池

2.5.1 创建全局连接池

from dbutils.pooled_db import PooledDB
import pymysql

POOL = PooledDB(
    creator=pymysql,
    maxconnections=10,      # 连接池最大连接数
    mincached=2,            # 启动时预建的最小空闲连接
    maxcached=6,            # 最大空闲连接数（0=无限制）
    blocking=True,          # 连接数满时是否阻塞等待
    maxusage=1000,          # 单个连接复用次数上限
    ping=1,                 # 取出连接时先 ping 保活
    host='127.0.0.1',
    port=3306,
    user='root',
    password='密码',
    database='数据库名',
    charset='utf8mb4'
)

dbutils提供数据库连接池，解决原生驱动每次请求新建 / 销毁连接的性能缺陷。

2.5.2 使用连接

import pymysql.cursors

def fetch_one(sql, params=None):
    conn = None
    cursor = None
    try:
        conn = POOL.connection()                          # 从池里取连接
        cursor = conn.cursor(pymysql.cursors.DictCursor)  # 字典游标
        cursor.execute(sql, params or ())
        return cursor.fetchone()
    except Exception as e:
        app.logger.error(f"DB error: {e}")
        return None
    finally:
        if cursor:
            cursor.close()
        if conn:
            conn.close()   # 不是真关闭，是把连接归还给池

2.5.3 对比

原生 pymysql	DBUtils 连接池
每次请求新建连接，频繁 TCP 握手，并发卡顿	连接复用，减少开销，高并发稳定
异常易出现连接泄漏，数据库连接数爆满	自动归还、自动检测断线重连
无连接上限控制，流量大时数据库崩溃	限制最大连接数，保护 MySQL

三.耗时版

3.1 接受任务并加入队列

@app.route("/boo", methods=["POST"])
def boo():
    # 1. 解析请求体为 JSON
    req_data = request.get_json()
    # 2. 获取 ordered_string 参数，默认空字符串
    ordered_string = req_data.get("ordered_string", "")
    # 3. 参数校验：不能为空
    if not ordered_string:
        return jsonify({"status": False, "error": "参数错误"}), 400

    # 4. 生成全局唯一任务 ID
    tid = str(uuid.uuid4())

    # 5. 创建 Redis 连接（每次请求新建，生产环境建议用连接池）
    redis_conn = redis.Redis(**REDIS_CONFIG)

    # 6. 构造任务字典，并序列化为 JSON 字符串
    task_info = json.dumps({
        'tid': tid,
        'ordered_string': ordered_string
    })

    # 7. 将任务推入 Redis 列表的左侧（队头）
    redis_conn.lpush("task_queue", task_info)

    # 8. 立即返回任务 ID，不等待实际处理
    return jsonify({
        "status": True,
        "data": {
            "tid": tid,
            "message": "任务已提交，等待处理"
        }
    })

UUID用于生成全局唯一标识符

为什么是 lpush？
将新任务推入 List 的左侧（头部），而后台 Worker 使用 brpop 从右侧（尾部）弹出。
效果：先入队的任务在 List 更靠右的位置，会被优先取出，从而保证 FIFO（先进先出）顺序。

3.2 worker处理

后台 Worker
一个独立的守护线程，启动时连接 Redis，无限循环地从 task_queue 阻塞式弹出任务（brpop），执行模拟的耗时操作（time.sleep(2)），然后将结果推入 result_queue。

初始化 Redis 连接并保持长连接，这条连接在整个 Worker 生命周期内只创建一次，并反复使用。

Redis作用：

一个基于内存的“数据结构服务器”，主要用来解决高并发下关系型数据库（如 MySQL）的读写瓶颈，提升系统响应速度。

def worker():
    """后台工作线程：从任务队列取任务，处理后放入结果队列"""
    try:
        redis_conn = redis.Redis(**REDIS_CONFIG)
        logging.info("Worker 启动成功，等待任务...")

        while True:
            # 1. 阻塞式从任务队列右侧弹出任务
            task_data = redis_conn.brpop("task_queue", timeout=1)

            if task_data:
                queue_name, task_json = task_data
                task = json.loads(task_json)
                tid = task.get('tid')
                ordered_string = task.get('ordered_string')

                logging.info(f"Worker 开始处理任务: {tid}")

                # 2. 模拟耗时处理
                time.sleep(2)
                result = f"处理完成: {ordered_string}"

                # 3. 将结果放入结果队列
                result_data = json.dumps({
                    'tid': tid,
                    'result': result,
                    'status': 'completed'
                })
                redis_conn.lpush("result_queue", result_data)

                logging.info(f"Worker 完成任务: {tid}")

    except Exception as e:
        logging.error(f"Worker 异常: {e}")

为什么用 brpop 而不是 rpop？

rpop 是无阻塞的，如果队列为空立即返回 None，这样 Worker 就会在 while True 中高速空转，白白消耗 CPU。

brpop 在无任务时让线程进入休眠，不占用 CPU，有新任务时立即唤醒，既高效又节省资源。

Worker 与 Web 接口的协作全景
客户端 → POST /boo（带 ordered_string）

Web 线程 → 生成 tid，lpush 到 task_queue，立即返回 tid

Worker 线程 → brpop 拿到任务，执行耗时操作，lpush 结果到 result_queue

客户端 → 轮询 GET /result/<tid>，遍历 result_queue 查找匹配的 tid，获取结果

四.附

1.Redis下载

Redis下载https://github.com/tporadowski/redis/releases

2.状态码

常见状态码
状态码	含义	使用场景
200	OK 请求成功	正常返回业务数据
400	Bad Request 参数非法	缺少参数、参数格式错误
401	Unauthorized 未认证	token 不存在、token 失效
403	Forbidden 权限不足	登录成功但无操作权限
404	Not Found 接口不存在	访问不存在的路由
500	Internal Server Error	代码报错、数据库异常

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/Xcy20060113/article/details/162666851

【python| 框架】flask速通