如何实现与路由签名无关的FastAPI装饰器式访问守卫
问题
我编写了一段FastAPI代码,当路由函数包含request参数时,role_guard装饰器可以正常运行:
@router.get("/me") @role_guard(UserRole.USER) async def read_user_me(request: Request, user: User = Depends(get_current_user)): return user.to_base()
但移除request参数后,装饰器立即失效。我希望避免函数中存在未使用的参数,同时将装饰器与路由逻辑解耦,想知道是否能通过装饰器注入所需的request参数?
ChatGPT生成了一段看似可行的代码,但实际无法运行:
def role_guard(role: UserRole): """ Minimum required role for that route. Also injects the user object into the route. """ def decorator(func): @wraps(func) async def wrapper(*args, **kwargs): request = kwargs.get('request') if not request: for arg in args: if isinstance(arg, Request): request = arg break if request: authorization: str = request.headers.get("Authorization") ## more logic here raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated", ) if 'request' not in inspect.signature(func).parameters: async def wrapper_with_request(request: Request, *args, **kwargs): return await wrapper(request=request, *args, **kwargs) return wrapper_with_request return wrapper return decorator
求可行的实现方式。
可行实现方案
FastAPI的路由参数由内置依赖注入系统管理,手动通过装饰器注入参数会破坏其参数解析逻辑,这也是ChatGPT生成代码失效的核心原因。以下是两种符合FastAPI设计逻辑的可行方案:
方案1:将角色校验改造为依赖项(推荐)
这是最贴合FastAPI设计理念的方式,完全解耦路由逻辑与权限校验:
from fastapi import Depends, Request, HTTPException, status from typing import Annotated from enum import Enum class UserRole(Enum): USER = "user" ADMIN = "admin" # 模拟获取当前用户的依赖逻辑 async def get_current_user(request: Request) -> User: token = request.headers.get("Authorization") if not token: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="未认证") # 替换为实际的用户查询逻辑 return User(id=1, role=UserRole.USER, name="test_user") # 角色校验依赖 async def role_guard(required_role: UserRole, user: User = Depends(get_current_user)): # 根据实际需求调整角色校验逻辑,这里假设枚举值可直接比较 if user.role != required_role: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="权限不足") return user # 路由使用示例:无需声明request参数 @router.get("/me") async def read_user_me(user: Annotated[User, Depends(lambda: role_guard(UserRole.USER))]): return user.to_base()
依赖项内部会自动通过FastAPI的依赖系统获取Request对象(已在get_current_user中声明),路由函数无需额外声明未使用的参数,完全实现解耦。
方案2:改造装饰器,通过FastAPI内置方式获取Request
若坚持使用装饰器形式,可通过FastAPI的Request依赖机制在装饰器内部获取请求对象,而非手动注入参数:
from fastapi import Request, HTTPException, status from functools import wraps import inspect from enum import Enum class UserRole(Enum): USER = "user" ADMIN = "admin" def role_guard(required_role: UserRole): def decorator(func): sig = inspect.signature(func) has_request_param = 'request' in sig.parameters @wraps(func) async def wrapper(*args, **kwargs): # 获取Request对象 request = None if has_request_param: # 从路由函数参数中提取request request = kwargs.get('request') or next((arg for arg in args if isinstance(arg, Request)), None) else: # 通过FastAPI的依赖系统自动注入Request from fastapi import Depends request = await Depends(Request)() # 模拟用户获取与角色校验逻辑 token = request.headers.get("Authorization") if not token: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="未认证") user = User(id=1, role=UserRole.USER, name="test_user") if user.role != required_role: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="权限不足") # 调用原路由函数,根据是否有request参数决定是否传递 if not has_request_param: return await func(*args, **kwargs) return await func(*args, **kwargs) return wrapper return decorator # 路由使用示例:无需声明request参数 @router.get("/me") @role_guard(UserRole.USER) async def read_user_me(user: User = Depends(get_current_user)): return user.to_base()
此方案虽然可行,但不如依赖项方案优雅,因为FastAPI的装饰器设计初衷是修改路由元数据或处理响应,依赖注入系统才是处理参数与权限校验的标准方式。
原代码失效原因
FastAPI在解析路由时,会根据函数签名自动注入对应参数(如Request、Depends声明的对象)。ChatGPT生成的代码尝试手动为包装函数添加request参数,但FastAPI无法识别这种手动添加的参数,导致无法注入有效的Request对象,最终装饰器因获取不到请求头而抛出未认证异常。
内容的提问来源于stack exchange,提问作者bb1
相关产品推荐
相关产品推荐

