backgroundtasks 不能直接 await，因其非协程而是任务注册器；需调用 add_task() 且不加 await，函数内自行处理异步操作；后台任务须新建数据库 session，避免复用请求级 session；它适合轻量、非关键任务，而 celery 更适合生产环境的可靠异步任务。
BackgroundTasks 为什么不能直接 await
FastAPI 的 BackgroundTasks 不是协程，它只是个任务注册器——你调用 add_task() 时，函数被放进队列，等主请求返回后才执行。如果你写 await background_tasks.add_task(send_email, user_id)，会报 TypeError: object BackgroundTasks can't be used in 'await' expression。
常见错误现象：把后台任务当异步函数用，结果路由卡住、超时，或者干脆启动失败。
- 正确做法是调用
background_tasks.add_task()，不加await - 传入的函数本身可以是普通函数，也可以是
async def函数；但 FastAPI 会用asyncio.to_thread()或事件循环调度，不保证并发安全 - 如果函数里用了
asyncio.sleep()或httpx.AsyncClient，必须确保它被正确 await（即函数内部自己处理），add_task()不帮你 await
后台任务里怎么安全用数据库连接
FastAPI 默认不会为你在后台任务中复用请求生命周期内的数据库 session（比如 SQLAlchemy 的 SessionLocal）。直接传 session 对象进去，大概率遇到 AttributeError: 'Session' object has no attribute 'execute' 或连接已关闭。
使用场景：发通知、写日志、异步导出报表——都需要查或写库，但又不能拖慢 HTTP 响应。
- 不要把 request-scoped 的
dbsession 直接传进add_task() - 应该在后台函数内部新建独立 session，例如：
with SessionLocal() as db: - 记得手动
db.commit()和db.close()，FastAPI 不接管这个生命周期 - 如果用的是异步 ORM（如 SQLModel + asyncpg），则必须用
async with AsyncSession() as db:，且函数得是async def
BackgroundTasks 和 Celery 到底该选谁
BackgroundTasks 是内存级、无持久化、无重试、无分布式能力的轻量机制；适合秒级完成、允许丢失、不关键的任务（比如埋点上报、缓存预热）。一旦进程重启，所有待执行任务就丢了。
性能与兼容性影响明显：它共享主线程事件循环，大量 CPU 密集型任务（如图像压缩）会阻塞整个 FastAPI 应用；而 Celery 配合 Redis/RabbitMQ 能真正解耦、限流、监控、重试。
- 选
BackgroundTasks：调试阶段、内部工具、低频非核心操作、不想引入新组件 - 选 Celery：生产环境、需保障送达率、任务耗时 > 1s、要分发到多台机器、需要定时/重试/优先级
- 混用也行：用
BackgroundTasks做快速响应层，再由它触发 Celery task，避免用户等待
后台任务报错会被吞掉，怎么捕获和调试
默认情况下，BackgroundTasks 中抛出的异常完全静默——HTTP 请求照常返回 200，但你的邮件没发出去、日志没写成，还找不到原因。
容易被忽略的地方：没有日志、没设 Sentry、也没做最简单的 try/except 包裹。
- 务必在后台函数开头加
try/except，至少记录logging.exception("Background task failed") - 别依赖全局异常处理器（
@app.exception_handler），它对BackgroundTasks无效 - 如果用 structlog 或 Sentry，确保初始化在函数内（或模块顶层），而不是只在中间件里初始化
- 本地调试时，可以把任务改成同步执行：临时替换
background_tasks.add_task(func)为func()，方便断点跟进
add_task()，而是任务失败时没人知道它失败了。大量免费API接口：立即使用
涵盖生活服务API、金融科技API、企业工商API、等相关的API接口服务。免费API接口可安全、合规地连接上下游，为数据API应用能力赋能！
