什么是 BackgroundTasks？
BackgroundTasks 是 FastAPI 提供的一个强大工具，它允许你将一些非紧急的、耗时的操作（例如发送邮件、处理数据、调用第三方 API 等）放到“后台”去执行，而不是让用户一直等待这些操作完成。
它的核心思想是：先快速返回响应给客户端，然后在后台慢慢处理其他任务。
为什么要使用它？
想象一下这些场景：
用户注册：用户点击注册后，你需要在数据库中创建用户记录，然后发送一封欢迎邮件。发送邮件可能需要几秒钟（与第三方邮件服务通信）。你不应该让用户盯着空白页面等待邮件发送成功才看到“注册成功”的提示。更好的体验是，立即告诉用户“注册成功”，然后在后台默默发送邮件。
文件上传处理：用户上传了一个大视频，你需要先保存文件，然后启动一个后台任务来转码、生成缩略图、分析内容等。这些操作非常耗时，应该立即响应“上传成功”，处理在后台进行。
写入日志或分析数据：记录用户的请求行为到数据库或日志文件，这个操作也不应该阻塞主响应。
在这些情况下，使用 BackgroundTasks 可以显著提升应用的响应速度和用户体验。
工作原理
FastAPI 的 BackgroundTasks 机制非常直观：
你将一个或多个函数（任务）添加到
BackgroundTasks对象中。FastAPI 会先将响应返回给客户端。
在当前请求的上下文结束后，FastAPI 会在同一个进程中依次执行你添加的所有后台任务。
重要特点：
同步执行：默认情况下，任务是同步执行的（一个接一个）。如果一个任务很慢，它会阻塞队列中的下一个任务。
与请求同进程：它适用于轻量级的后台任务。对于非常耗时或CPU密集型的任务，最好使用更强大的工具（如
Celery,RQ,ARQ），因为它们可以在单独的进程或 worker 中运行，避免阻塞你的主应用。简单易用：无需额外的基础设施（如 Redis、RabbitMQ），开箱即用。
如何使用？
1. 基本使用步骤
a. 导入 BackgroundTasks
b. 在路径操作函数的参数中声明它
c. 编写要后台运行的函数
d. 使用 add_task() 方法添加任务
e. 返回响应
from fastapi import FastAPI, BackgroundTasks
import timeapp = FastAPI()# 这是一个模拟的耗时任务，例如发送邮件
def send_notification(email: str, message: str):print(f"开始发送邮件给 {email}...")time.sleep(3) # 模拟耗时操作，比如网络请求print(f"邮件已发送给 {email}：{message}")# 这里可以是实际的发送邮件逻辑，如使用 smtplib@app.post("/register/{username}")
async def register_user(username: str, email: str, background_tasks: BackgroundTasks):# 1. 首先执行主要的快速逻辑（例如将用户保存到数据库）# user = create_user_in_db(username, email) <-- 伪代码print(f"用户 {username} 已成功创建于数据库。")# 2. 将耗时任务添加到后台background_tasks.add_task(send_notification, # 要执行的函数email, # 传递给函数的第一个参数f"嗨，{username}！欢迎加入我们！" # 传递给函数的第二个参数)# 3. 立即返回响应return {"message": "用户注册成功","username": username,"email": email}依赖注入与路径操作函数
你可以将 BackgroundTasks 不仅注入到路径操作函数中，还可以注入到依赖项（Dependencies） 中，这使得代码更加模块化和可复用。
from fastapi import Dependsdef write_log(message: str):with open("log.txt", mode="a") as log:log.write(message + "\n")def get_query(background_tasks: BackgroundTasks, q: str | None = None):if q:# 在依赖项中添加后台任务background_tasks.add_task(write_log, f"查询参数: {q}")return q@app.get("/items/")
async def read_items(query: str = Depends(get_query)):# 这个路径操作函数本身可能也会添加任务# background_tasks.add_task(...)return {"query": query}
