如何以Pythonic方式（如装饰器）实现Flask API统一RabbitMQ RPC授权？

嘿，这个需求太典型了！用装饰器绝对是最Pythonic的解决方案，完美适配你用Flask + flask-restful构建API的场景，还能轻松实现所有接口的统一授权校验。我结合你的业务场景（Bearer Token授权、RPC调用RabbitMQ的request_authz方法），给你一步步拆解实现思路和代码示例：

1. 先明确request_authz的调用逻辑（适配你的业务场景） #

首先咱先对齐request_authz的预期行为：它需要从请求中提取Bearer Token，通过pika调用RabbitMQ的RPC服务校验权限，授权成功返回用户信息，失败则抛出对应异常。这里我先补全基础逻辑（你可以根据实际的RabbitMQ交互代码调整）：

import pika
from flask import request

def request_authz():
    """从请求中提取Bearer Token，通过RabbitMQ RPC完成授权校验"""
    # 提取并校验Authorization头
    auth_header = request.headers.get('Authorization')
    if not auth_header or not auth_header.startswith('Bearer '):
        raise ValueError("无效的授权头，请提供Bearer Token")
    
    token = auth_header.split(' ')[1]
    
    # 这里替换成你实际的pika操作RabbitMQ RPC的代码
    # 示例：发送Token到RabbitMQ，等待授权响应
    auth_result = call_rabbitmq_rpc(token)  # 假设该函数返回授权结果字典
    
    if not auth_result.get('authorized'):
        raise PermissionError("Token无效或用户权限不足")
    
    # 授权成功返回用户信息，方便后续接口使用
    return auth_result.get('user_info')

2. 实现通用的授权装饰器 #

接下来写一个require_auth装饰器，它会自动在接口处理前调用request_authz，统一处理授权失败的情况，返回标准HTTP错误响应：

from flask import jsonify, g
from functools import wraps

def require_auth(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        try:
            # 执行授权校验，成功则将用户信息存入g对象（全局可访问）
            user_info = request_authz()
            g.user = user_info
        except ValueError as e:
            return jsonify({"error": str(e)}), 401  # 未授权状态码
        except PermissionError as e:
            return jsonify({"error": str(e)}), 403  # 权限不足状态码
        # 授权通过，执行原接口逻辑
        return f(*args, **kwargs)
    return decorated_function

3. 在flask-restful的Resource中使用装饰器 #

针对你需要实现的POST API（获取metadata参数），我们可以用两种方式给Resource接口加授权：

方式一：给单个方法加装饰器（适合部分方法需要授权的场景） #

from flask_restful import Resource, reqparse

class MetadataHandler(Resource):
    # 定义POST请求的参数解析规则
    parser = reqparse.RequestParser()
    parser.add_argument(
        'metadata', 
        type=dict, 
        required=True, 
        help="metadata参数不能为空，请传入合法的JSON对象"
    )

    @require_auth
    def post(self):
        # 解析并获取metadata参数
        args = self.parser.parse_args()
        metadata = args['metadata']
        
        # 这里写你的业务逻辑：比如处理metadata、调用RabbitMQ等
        return jsonify({
            "status": "success",
            "message": "metadata已接收",
            "received_metadata": metadata,
            "authorized_user": g.user
        })

方式二：给整个Resource类加统一装饰器（适合所有方法都需要授权的场景） #

如果这个Resource的所有接口（比如GET/PUT/DELETE）都需要授权，直接用method_decorators属性统一配置即可：

class MetadataHandler(Resource):
    # 给类下所有方法统一添加授权装饰器
    method_decorators = [require_auth]
    
    parser = reqparse.RequestParser()
    parser.add_argument(
        'metadata', 
        type=dict, 
        required=True, 
        help="metadata参数不能为空，请传入合法的JSON对象"
    )

    def post(self):
        args = self.parser.parse_args()
        metadata = args['metadata']
        return jsonify({
            "status": "success",
            "received_metadata": metadata,
            "authorized_user": g.user
        })

    def get(self):
        # 这个GET接口也会自动经过授权校验
        return jsonify({
            "message": "这是受保护的GET接口",
            "user": g.user
        })

4. 额外优化：全局授权拦截（可选） #

如果你的所有API都必须经过授权，不想逐个添加装饰器，可以用Flask的before_request钩子实现全局拦截（注意排除登录、健康检查等无需授权的接口）：

@app.before_request
def global_auth_check():
    # 排除不需要授权的接口路径
    excluded_paths = ['/api/login', '/api/health']
    if request.path not in excluded_paths:
        try:
            user_info = request_authz()
            g.user = user_info
        except ValueError as e:
            return jsonify({"error": str(e)}), 401
        except PermissionError as e:
            return jsonify({"error": str(e)}), 403

总结 #

• 装饰器方式更灵活，适合部分接口需要授权的场景；
• 全局钩子适合所有接口强制授权的场景；
• 两种方案都能完美结合你的request_authz方法，实现统一的授权校验逻辑。

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