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 BackgroundTaskslogger = 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）对 asy
ncio task 异常无效，除非你手动调用
task.exception()
进阶：封装带异常监听的 BackgroundTask 类
如果你项目中大量使用 background task 且需要统一监控，可以轻量封装一个替代方案：
- 用
asyncio.create_task()显式创建 task，并保存引用 - 在 task 完成后主动调用
task.exception()检查是否出错 - 把异常转发到日志或告警通道（注意别阻塞 event loop）
关键点在于：不能只靠 add_task，得自己管理 task 生命周期。FastAPI 原生 BackgroundTasks 就是“fire-and-forget”，真要可靠，就得自己多走一步。
