如何基于SQLAlchemy Core实现@token_required装饰器保护用户详情接口？

问题场景 #

已实现登录接口（通过用户名密码验证后返回包含用户uid的JWT），现在需要为通过uid获取用户详情的接口添加@token_required验证功能，使用的是SQLAlchemy Core而非ORM，无法参考常规ORM资料推进。

解决方案 #

1. 编写token_required装饰器

这个装饰器会从请求头提取Token，验证合法性后用SQLAlchemy Core查询用户信息，再将用户数据传递给视图函数：

import jwt
from flask import request, make_response, jsonify
from sqlalchemy import text
from your_app_module import engine  # 替换为你的engine实际导入路径

def token_required(f):
    def decorated(*args, **kwargs):
        token = None
        # 从请求头获取Authorization字段，格式为Bearer <token>
        if 'Authorization' in request.headers:
            auth_parts = request.headers['Authorization'].split(" ")
            token = auth_parts[1] if len(auth_parts) > 1 else None

        if not token:
            return make_response(jsonify({'message': 'Token缺失！'}), 401)

        try:
            # 解码Token拿到用户uid
            decoded_data = jwt.decode(token, app.config['SECRET_KEY'], algorithms=["HS256"])
            current_uid = decoded_data['uid']
            
            # 用SQLAlchemy Core查询用户信息
            conn = engine.connect()
            user_result = conn.execute(
                text("SELECT id, user_name, email_address, dob, uid FROM user WHERE uid = :uid"),
                {"uid": current_uid}
            ).fetchone()
            
            if not user_result:
                return make_response(jsonify({'message': '用户不存在！'}), 401)
            
            current_user = dict(user_result)
        except jwt.ExpiredSignatureError:
            return make_response(jsonify({'message': 'Token已过期！'}), 401)
        except jwt.InvalidTokenError:
            return make_response(jsonify({'message': '无效Token！'}), 401)
        except Exception as e:
            return make_response(jsonify({'message': f'错误：{str(e)}'}), 500)
        finally:
            conn.close()  # 确保数据库连接关闭

        # 把用户数据传入原视图函数
        return f(current_user, *args, **kwargs)
    return decorated

2. 修改用户详情路由

添加@token_required装饰器，同时修复原SQL语句的参数绑定错误：

@app.route('/users/<uid>', methods=['GET'])
@token_required
def profile_view(current_user, uid):
    # 可选：校验请求的uid与Token中的uid一致，防止越权访问
    if current_user['uid'] != uid:
        return make_response(jsonify({'message': '无权限访问该用户信息！'}), 403)
    
    conn = engine.connect()
    str_sql = text("SELECT * FROM user WHERE uid = :uid")
    results = conn.execute(str_sql, {"uid": uid}).fetchall()
    conn.close()
    return users_scehma.dump(results)

3. 关键注意点

• Token传递：前端请求时需在Authorization请求头中携带Bearer <你的Token>格式内容
• 越权防护：通过校验请求uid与Token中uid的一致性，避免用户访问他人信息
• 连接管理：每次数据库操作后手动关闭连接，避免资源泄漏
• 异常覆盖：处理了Token缺失、过期、无效及用户不存在等常见异常场景

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