Flask 2中如何将Auth Middleware生成的数据传递给Blueprint函数

可以通过WSGI environ 字典实现请求级数据传递，也可以结合Flask的g对象优化使用体验，两种方案都不需要session或数据库，且能保证多请求之间的数据隔离。

方案一：直接使用WSGI environ传递 #

1. 修改中间件，将数据存入environ #

在my_middleware.py的验证通过逻辑后，把my_data存入environ，注意使用带自定义前缀的键名，避免与框架内置键冲突：

from werkzeug.wrappers import Request, Response

class MiddlewareExample:
    def __init__(self, app):
        self.app = app

    def __call__(self, environ, start_response):
        request = Request(environ)
        headers = request.headers

        # 省略验证逻辑...
        result_validation = some_validations(headers)
        my_data = result_validation['some_result']

        if my_data['some_error']:
            # 返回错误响应逻辑...
            response = Response("验证失败", status=401)
            return response(environ, start_response)

        # 将数据存入environ，用自定义前缀避免冲突
        environ['my_app.my_data'] = my_data

        return self.app(environ, start_response)

2. 在Blueprint视图中读取数据 #

在my_custom_blueprint.py中，通过request.environ获取传递的数据：

from flask import request, Blueprint

demo_routes = Blueprint('demo', __name__, url_prefix='/demo')

@demo_routes.route('/', methods=['GET'])
def foo():
    # 从environ中获取数据
    my_data = request.environ.get('my_app.my_data')
    if not my_data:
        return "数据未正常传递", 400
    
    # 后续使用my_data的业务逻辑
    return f"获取到验证数据: {my_data}"

方案二：结合Flask的g对象传递（更符合Flask开发习惯） #

g是Flask提供的请求上下文对象，专门用于存储请求级别的临时数据，适合在多个视图或辅助函数中共享数据。

1. 中间件存入environ（同方案一） #

保留方案一中中间件写入environ的逻辑。

2. 注册全局before_request钩子，将数据同步到g #

在app.py中添加钩子，把environ中的数据转移到g对象：

from flask import Flask, g, request
from my_custom_blueprint import demo_routes as my_blueprint

def create_app():
    app = Flask(__name__)

    # 注册钩子，将environ中的数据同步到g
    @app.before_request
    def load_request_data():
        g.my_data = request.environ.get('my_app.my_data')

    with app.app_context():
        app.register_blueprint(my_blueprint)
        app.wsgi_app = MiddlewareExample(app.wsgi_app)
        return app

3. 在视图中通过g读取数据 #

修改my_custom_blueprint.py，直接从g对象获取数据：

from flask import g, Blueprint

demo_routes = Blueprint('demo', __name__, url_prefix='/demo')

@demo_routes.route('/', methods=['GET'])
def foo():
    my_data = g.my_data
    if not my_data:
        return "数据未正常传递", 400
    
    # 业务逻辑处理
    return f"通过g对象获取数据: {my_data}"

注意事项 #

• 务必使用带自定义前缀的键名存储到environ，比如my_app.my_data，避免与Werkzeug或其他扩展的内置键冲突。
• 两种方案都是请求隔离的，每个请求的my_data独立存储，不会出现多请求数据混乱的问题。

内容的提问来源于stack exchange，提问作者Aroldo Landeros