FastAPI内部调用带会话校验装饰器的接口时,如何避免重复会话检查?
FastAPI内部调用带会话校验装饰器的接口时,如何避免重复会话检查?
我完全懂你遇到的麻烦——自己写的会话校验装饰器在接口内部调用其他带装饰的接口时,会重复触发两次会话检查,而你又不想把每个接口都拆成路由层和实现层,确实挺闹心的。下面给你几个实用又优雅的解决思路,你可以根据自己的代码风格挑:
方案一:给装饰器加「跳过校验」的开关(最小改动)
这种方法不用大改现有代码结构,只给require_session装饰器加一个可选参数,允许内部调用时手动跳过校验。
先修改你的装饰器:
from functools import wraps import fastapi as fa import exc # 你的异常模块 def require_session(skip_check: bool = False): # 带参数的装饰器需要多一层嵌套,同时兼容无参数直接使用的情况 def decorator(func): @wraps(func) async def wrapper(*args, **kwargs): # 优先检查是否需要跳过校验 if skip_check: return await func(*args, **kwargs) # 原有的会话校验逻辑 request = kwargs.get('request') if not request: raise exc.access_denied() session_key = request.headers.get('X-Session-Key') if session_key and Session.check(session_key): return await func(*args, **kwargs) raise exc.access_denied() return wrapper # 兼容@require_session直接使用的场景(不带参数) if callable(skip_check): func = skip_check skip_check = False return decorator(func) return decorator
然后内部调用时,用带开关的版本包装原函数:
@administrators_router.post('') @require_session async def create_administrator(request: fa.Request, data: AdministratorSchema) -> AdministratorDBSchema: data_dump = data.model_dump() if isinstance(data_dump['employee'], int): # 内部调用时手动跳过校验 await require_session(skip_check=True)(get_employee)(request, data_dump['employee']) try: admin = Administrator.create(**data_dump) except pw.IntegrityError as e: raise exc.integrity(e) return admin
方案二:用Request.state标记已校验(最省心)
利用FastAPI的request.state对象存储一个「会话已校验」的标记,装饰器先检查这个标记,已校验过就直接放行。
修改require_session装饰器:
def require_session(func): @wraps(func) async def wrapper(*args, **kwargs): request = kwargs.get('request') # 先检查请求状态里的校验标记 if hasattr(request.state, 'session_verified') and request.state.session_verified: return await func(*args, **kwargs) # 原有的会话校验逻辑 session_key = request.headers.get('X-Session-Key') if session_key and Session.check(session_key): # 标记会话已校验,后续调用直接跳过 request.state.session_verified = True return await func(*args, **kwargs) raise exc.access_denied() return wrapper
这个方案几乎不需要改业务代码,只要装饰器逻辑加几行就行。当第一个带装饰器的接口被调用时,会在request.state上打标记,后续内部调用任何带require_session的接口时,都会自动跳过重复校验,完美适配你的现有代码!
方案三:改用FastAPI Depends依赖注入(最佳实践)
其实FastAPI的依赖注入系统就是为了解决这类重复校验问题而设计的,比自定义装饰器更灵活,也更符合框架的设计理念。
首先把会话校验写成一个依赖项:
async def verify_session(request: fa.Request): session_key = request.headers.get('X-Session-Key') if not session_key or not Session.check(session_key): raise exc.access_denied() # 可以返回会话对象,后续业务逻辑需要时直接用 return Session.check(session_key)
然后在路由里用Depends代替装饰器:
# 员工路由 @employees_router.get('/{id}') async def get_employee( request: fa.Request, id: int, # 自动注入会话校验 _=fa.Depends(verify_session) ) -> EmployeeDBSchema: employee = Employee.get_or_none(id=id) if employee is None: raise exc.instance_by_field_not_found(Employee, id) return employee # 管理员路由 @administrators_router.post('') async def create_administrator( request: fa.Request, data: AdministratorSchema, _=fa.Depends(verify_session) ) -> AdministratorDBSchema: data_dump = data.model_dump() if isinstance(data_dump['employee'], int): # 内部调用时直接传参,不需要处理校验(当前路由已经做过了) await get_employee(request, data_dump['employee'], _=None) try: admin = Administrator.create(**data_dump) except pw.IntegrityError as e: raise exc.integrity(e) return admin
这种方法的好处是后续扩展更方便——比如要在业务逻辑里用会话信息,直接把依赖返回的会话对象作为参数就行,不用再从请求里取。
备注:内容来源于stack exchange,提问作者kos1posha
相关产品推荐
相关产品推荐

