PAYATHON 2026

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") 只会输出文本,不会验证会话。实际使用时需要在装饰器中实现检查逻辑。

备注:内容仅供参考。