用上下文管理器+装饰器优雅管理Flask请求生命周期，告别冗余代码

上一篇文章我们聊了Python上下文管理器的核心用法，有读者留言："能不能结合Flask讲讲实际项目中的最佳实践？" 今天就来揭秘如何用「上下文管理器+装饰器」组合拳，优雅管理Flask的请求-响应生命周期，解决实际开发中的痛点问题。

一、为什么需要管理请求生命周期？

先看一个典型的Flask视图函数：

@app.route('/api/order', methods=['POST'])
def create_order():
    # 1. 初始化资源
    request_id = str(uuid.uuid4())
    db_conn = get_db_connection()
    logger = get_logger()
    start_time = time.time()
    
    try:
        # 2. 业务逻辑
        logger.info(f"[{request_id}] 开始处理订单")
        data = request.get_json()
        order_id = db_conn.execute("INSERT INTO orders...", data).lastrowid
        db_conn.commit()
        return jsonify({"order_id": order_id, "request_id": request_id})
    except Exception as e:
        # 3. 异常处理
        db_conn.rollback()
        logger.error(f"[{request_id}] 订单处理失败: {str(e)}")
        return jsonify({"error": "处理失败"}), 500
    finally:
        # 4. 清理资源
        db_conn.close()
        logger.info(f"[{request_id}] 请求耗时: {time.time()-start_time:.3f}s")

这段代码有三个明显问题：

1. 冗余代码：每个视图函数都要写资源初始化、清理、日志记录的重复代码
2. 耦合严重：业务逻辑和横切关注点（日志、事务）混在一起
3. 维护困难：修改日志格式或事务逻辑需要改动所有视图函数

而用「上下文管理器+装饰器」能完美解决这些问题！

二、核心实现：请求上下文装饰器

2.1 定义请求上下文管理器

首先实现一个管理请求生命周期的上下文管理器，封装资源的初始化与清理：

import uuid
import time
from contextlib import contextmanager
from flask import g, request
import logging

@contextmanager
def request_context():
    # 1. 请求开始阶段：初始化资源
    g.request_id = str(uuid.uuid4())  # 用Flask的g对象存储上下文信息
    g.start_time = time.time()
    g.logger = logging.getLogger(f"request:{g.request_id}")
    
    # 初始化数据库连接
    g.db_conn = get_db_connection()
    g.logger.info(f"开始处理请求: {request.path}")
    
    try:
        yield# 2. 中间阶段：执行视图函数业务逻辑
    except Exception as e:
        # 3. 异常处理阶段
        g.db_conn.rollback()
        g.logger.error(f"请求处理异常: {str(e)}", exc_info=True)
        raise# 继续抛出异常让Flask统一处理
    finally:
        # 4. 请求结束阶段：清理资源
        g.db_conn.close()
       耗时 = time.time() - g.start_time
        g.logger.info(f"请求处理完成，耗时: {耗时:.3f}s")

2.2 结合装饰器注入上下文

用装饰器为视图函数自动注入上述上下文管理逻辑：

from functools import wraps
from flask import jsonify

def with_request_context(f):
    @wraps(f)  # 保留原函数元信息
    def wrapper(*args, **kwargs):
        with request_context():  # 调用上下文管理器
            try:
                return f(*args, **kwargs)  # 执行视图函数
            except Exception as e:
                # 统一异常响应格式
                return jsonify({
                    "error": str(e),
                    "request_id": g.request_id  # 包含request_id方便排查问题
                }), 500
    return wrapper

2.3 改造后的视图函数

用装饰器标记需要管理生命周期的视图，代码瞬间清爽：

@app.route('/api/order', methods=['POST'])
@with_request_context  # 注入请求生命周期管理
def create_order():
    # 直接使用上下文管理器中初始化的资源
    g.logger.info("处理订单创建请求")
    
    data = request.get_json()
    cursor = g.db_conn.execute(
        "INSERT INTO orders (user_id, amount) VALUES (%s, %s)",
        (data['user_id'], data['amount'])
    )
    g.db_conn.commit()
    
    return jsonify({
        "order_id": cursor.lastrowid,
        "request_id": g.request_id  # 返回request_id方便追踪
    })

三、进阶功能：扩展生命周期管理

3.1 支持自定义配置

让装饰器支持参数，灵活控制生命周期行为：

def with_request_context(need_db=True, log_level=logging.INFO):
    def decorator(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            with request_context(need_db=need_db, log_level=log_level):
                # 同上...
        return wrapper
    return decorator

# 使用示例：不需要数据库的视图
@app.route('/api/health')
@with_request_context(need_db=False)
def health_check():
    return jsonify(status="healthy")

3.2 实现事务自动管理

在上下文管理器中增强数据库事务支持：

@contextmanager
def request_context(need_db=True):
    # ... 省略初始化代码
    
    if need_db:
        g.db_conn = get_db_connection()
        g.transaction = g.db_conn.begin()  # 开启事务
    try:
        yield
        if need_db:
            g.transaction.commit()  # 无异常则提交
    except:
        if need_db:
            g.transaction.rollback()  # 异常则回滚
        raise
    finally:
        if need_db:
            g.db_conn.close()

3.3 跨函数共享上下文信息

在工具函数中直接访问上下文信息，无需参数传递：

# 工具函数：发送通知
def send_notification(user_id, message):
    # 直接从g对象获取当前请求上下文
    logger = g.logger
    logger.info(f"向用户{user_id}发送通知")
    
    # 实际发送逻辑...

# 在视图函数中调用
@app.route('/api/order', methods=['POST'])
@with_request_context
def create_order():
    # ... 订单创建逻辑
    send_notification(data['user_id'], "订单创建成功")  # 无需传递logger
    return ...

四、生产级最佳实践

4.1 多层上下文组合

用ExitStack实现多上下文组合，满足复杂场景：

from contextlib import ExitStack

@contextmanager
def complex_request_context():
    with ExitStack() as stack:
        # 组合多个上下文
        stack.enter_context(request_context())  # 请求基础上下文
        stack.enter_context(redis_context())    # Redis连接上下文
        stack.enter_context(cache_context())    # 缓存上下文
        yield

4.2 与Flask扩展的集成

结合flask-sqlalchemy等扩展时的适配方案：

from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()

@contextmanager
def sa_request_context():
    try:
        yield
        db.session.commit()
    except:
        db.session.rollback()
        raise
    finally:
        db.session.remove()  # 归还连接到池

4.3 异步场景支持

在Flask异步视图中使用异步上下文管理器：

import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
asyncdef async_request_context():
    g.request_id = str(uuid.uuid4())
    g.start_time = time.time()
    # 异步初始化资源
    g.async_db = await get_async_db_connection()
    try:
        yield
    finally:
        await g.async_db.close()

# 异步装饰器
def with_async_context(f):
    @wraps(f)
    asyncdef wrapper(*args, **kwargs):
        asyncwith async_request_context():
            returnawait f(*args, **kwargs)
    return wrapper

# 异步视图
@app.route('/api/async-order', methods=['POST'])
@with_async_context
asyncdef async_create_order():
    # 异步处理逻辑...

五、避坑指南

5.1 不要在上下文外访问g对象 ：

# 错误示例：在非请求上下文调用
def bad_function():
    print(g.request_id)  # 会抛出RuntimeError

5.2 注意装饰器顺序 ：

# 正确顺序：路由装饰器在最外层
@app.route('/api/order')
@with_request_context
@validate_request  # 自定义参数校验装饰器
def create_order():
    ...

5.3 避免长时间占用资源 ：

上下文管理器会在请求结束后释放资源，避免在视图函数中执行耗时过长的操作。