FastAPI 异步路由可以使用同步装饰器吗?
数据阿明
结论
异步函数可以添加装饰器,但示例中的 admin_required 是同步包装器。调用 async def get_review() 后,它返回的是 coroutine,而包装器没有等待 coroutine 完成。FastAPI 可能因此将包装后的函数识别为同步路由,在工作线程中调用它,结果拿到的仍是 coroutine,导致响应处理失败或出现序列化错误。
装饰异步路由时,应使用异步包装器,并通过 await 调用原函数。
推荐写法
from functools import wraps
def admin_required(admin_function):
@wraps(admin_function)
async def check_admin(*args, **kwargs):
print("Verifiy admin session")
return await admin_function(*args, **kwargs)
return check_admin
路由的装饰顺序可以保持不变:
@review_router.get("/review")
@admin_required
async def get_review():
return "GET"
Python 会先应用靠近函数定义的 @admin_required,再将包装后的函数传给 @review_router.get 注册。@wraps 会保留原函数的元数据。由于包装函数本身使用 async def 定义,FastAPI 可以按异步路由处理它。
如果权限检查本身是异步的
如果检查管理员身份需要访问异步数据库或网络接口,也应在包装器中等待检查结果:
from functools import wraps
def admin_required(admin_function):
@wraps(admin_function)
async def check_admin(*args, **kwargs):
is_admin = await verify_admin_session()
if not is_admin:
raise PermissionError("Admin access required")
return await admin_function(*args, **kwargs)
return check_admin
verify_admin_session() 是用于说明调用位置的示例名称。实际的检查逻辑和未授权时的处理方式应根据应用实现调整。FastAPI 应用通常使用 HTTPException 返回 HTTP 错误,具体状态码则按接口约定选择。
注意事项
- 同步包装器直接返回异步函数的调用结果时,得到的是 coroutine。仅写
return admin_function(...)无法让异步函数运行完成。 - 上面的异步装饰器适用于异步函数。如果还要支持普通同步函数,需要根据函数类型分别处理,避免等待同步函数。
- 示例中的
print("Verifiy admin session")只会输出文本,不会验证会话。实际使用时需要在装饰器中实现检查逻辑。
备注:内容仅供参考。
