Flask RESTful API动态端点用户ID权限控制问题咨询

作为刚接触Flask两周就自研API的开发者，你已经搞定CRUD操作，这一步已经很扎实了～接下来只需要把用户认证和权限校验逻辑结合起来，就能实现John只能操作/api/john的需求。

我给你拆解下具体步骤和代码示例：

1. 先搞定用户认证（API场景推荐用JWT） #

因为你的是API服务，无状态的JWT认证会比Session更合适。我们用Flask-JWT-Extended这个扩展来快速实现：

安装依赖 #

pip install flask-jwt-extended

初始化配置 #

from flask import Flask, jsonify, request
from flask_jwt_extended import (
    JWTManager, create_access_token,
    jwt_required, get_jwt_identity
)

app = Flask(__name__)
# 设置JWT密钥（生产环境要存在环境变量里，别硬编码）
app.config["JWT_SECRET_KEY"] = "your-super-secret-key-here"
jwt = JWTManager(app)

# 模拟用户数据库（实际项目用SQLAlchemy等ORM连接真实数据库）
users = {
    "john": {"password": "john123", "tasks_count": 0},
    "jane": {"password": "jane123", "tasks_count": 0}
}

写登录接口生成Token #

@app.route("/login", methods=["POST"])
def login():
    username = request.json.get("username")
    password = request.json.get("password")
    
    if username not in users or users[username]["password"] != password:
        return jsonify({"msg": "用户名或密码错误"}), 401
    
    # 生成包含用户名的JWT Token
    access_token = create_access_token(identity=username)
    return jsonify(access_token=access_token), 200

2. 实现权限校验逻辑 #

核心思路：已认证用户操作时，必须确保URL中的UserID和当前登录用户的ID一致。我们可以在需要权限的路由里直接校验，也可以写个自定义装饰器复用逻辑。

路由实现（结合你的需求） #

# 未认证用户可访问：查看指定用户的指标
@app.route("/api/<user_id>", methods=["GET"])
def get_user_tasks(user_id):
    if user_id not in users:
        return jsonify({"msg": "用户不存在"}), 404
    return jsonify({"user_id": user_id, "tasks_count": users[user_id]["tasks_count"]}), 200

# 已认证用户才能操作，且只能操作自己的指标
@app.route("/api/<user_id>", methods=["POST", "PUT"])
@jwt_required()  # 这个装饰器会校验请求头里的Token
def increase_tasks(user_id):
    current_user = get_jwt_identity()  # 获取当前登录的用户名
    
    # 权限校验：当前用户只能操作自己的资源
    if current_user != user_id:
        return jsonify({"msg": "无权操作该用户的任务"}), 403
    
    # 执行计数增加逻辑（这里模拟+1）
    users[current_user]["tasks_count"] += 1
    return jsonify({"user_id": current_user, "tasks_count": users[current_user]["tasks_count"]}), 200

@app.route("/api/<user_id>", methods=["DELETE"])
@jwt_required()
def decrease_tasks(user_id):
    current_user = get_jwt_identity()
    
    if current_user != user_id:
        return jsonify({"msg": "无权操作该用户的任务"}), 403
    
    # 执行计数减少逻辑（模拟-1，注意边界值判断）
    if users[current_user]["tasks_count"] > 0:
        users[current_user]["tasks_count"] -= 1
    return jsonify({"user_id": current_user, "tasks_count": users[current_user]["tasks_count"]}), 200

# 重置任务计数的路由（同理）
@app.route("/api/<user_id>/reset", methods=["POST"])
@jwt_required()
def reset_tasks(user_id):
    current_user = get_jwt_identity()
    
    if current_user != user_id:
        return jsonify({"msg": "无权操作该用户的任务"}), 403
    
    users[current_user]["tasks_count"] = 0
    return jsonify({"user_id": current_user, "tasks_count": 0}), 200

3. 额外注意事项 #

• 数据库层面的优化：实际项目中，用户的任务计数应该存在数据库的用户表或关联表中，用ORM（比如Flask-SQLAlchemy）来操作会更安全可靠。
• 错误处理：可以统一处理JWT过期、无效等错误，返回更友好的提示。
• 密码安全：绝对不能明文存储密码！要使用werkzeug.security的generate_password_hash和check_password_hash来加密存储和校验密码。
• 生产环境配置：JWT密钥、数据库连接字符串等敏感信息要放在环境变量里，不要硬编码到代码中。

这样配置后，John登录拿到Token后，只能对/api/john发起POST/PUT/DELETE请求，访问其他用户的路径会返回403；未认证用户则只能通过GET请求查看任意用户的指标，完全符合你的需求～

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