fastapi BackgroundTasks 如何在请求结束后捕获异常

FastAPI的BackgroundTasks异常默认被事件循环丢弃,应在任务函数内用try/except捕获并记录日志;不推荐依赖shutdown钩子或全局异常处理器;需自行封装带异常监听的Task类以实现可靠监控。

BackgroundTasks 的异常默认会被吞掉

FastAPI 的 BackgroundTasks 在请求返回后异步执行,但它的底层是通过 asyncio.create_task() 启动协程,而这些任务一旦抛出未捕获异常,会直接被事件循环丢弃——你既收不到报错日志,也无法在请求上下文中感知失败。

这不是 Bug,是 asyncio 的默认行为:未被 await 的 task 异常不会传播,也不会中断主线程。

用 try/except 包裹 background task 函数体

最直接、最可控的方式是在你注册的 background 函数内部做异常兜底:

  • 所有可能出错的逻辑必须放在 try 块里
  • except 中记录日志(推荐用 logging.exception() 保留 traceback)
  • 避免在 background 函数里 raise 新异常,否则照样被吞
import logging
from fastapi import BackgroundTasks

logger = logging.getLogger(name)

def send_email_async(user_id: int): try:

模拟可能失败的操作

    raise ValueError("SMTP connection failed")

except Exception: logger.exception("Failed to send email for user %s", user_id)

@app.post("/notify") def notify_user(background_tasks: BackgroundTasks): background_tasks.add_task(send_email_async, user_id=123) return {"status": "accepted"}

不建议依赖 on_shutdown 或 global exception handler

有人想用 app.on_event("shutdown") 等待所有 background tasks 结束再检查异常,这不可行:

  • FastAPI 不提供对 background task 集合的引用,无法遍历或 await 它们
  • shutdown 时 event loop 可能已关闭,await task 会报 RuntimeError: Event loop is closed
  • 全局异常钩子(如 sys.excepthook)对 asyncio task 异常无效,除非你手动调用 task.exception()

进阶:封装带异常监听的 BackgroundTask 类

如果你项目中大量使用 background task 且需要统一监控,可以轻量封装一个替代方案:

  • asyncio.create_task() 显式创建 task,并保存引用
  • 在 task 完成后主动调用 task.exception() 检查是否出错
  • 把异常转发到日志或告警通道(注意别阻塞 event loop)

关键点在于:不能只靠 add_task,得自己管理 task 生命周期。FastAPI 原生 BackgroundTasks 就是“fire-and-forget”,真要可靠,就得自己多走一步。