FastAPI 后台任务 BackgroundTasks 完整实践指南：响应先行，处理后置

本文基于 FastAPI 官方文档《Hintergrundtasks》（背景任务教程）编写，围绕 BackgroundTasks 的核心用法展开：如何在响应发送之后再执行耗时操作（如发邮件、处理文件）、如何把任务函数挂到任务对象上、如何结合依赖注入在多层级共享同一个任务集合，以及其底层来自 Starlette 的实现细节。读完后，你将能够熟练地在 FastAPI 应用中实现"响应先行、处理后置"的异步后台处理方案。

什么是背景任务（BackgroundTasks）

背景任务是指在响应（Response）返回给客户端之后才执行的任务。这类操作有一个共同特征：Request 处理完后必须执行，但客户端不需要等待其完成就能收到响应。

官方文档给出的典型场景包括：

• 发送电子邮件通知：在某个动作执行后需要发送通知。由于连接邮件服务器并发送邮件通常比较"慢"（耗时数秒），可以先立即返回响应，把邮件放到后台发送；
• 数据处理：假设收到一个文件，它需要经过一个耗时的处理流程。可以先返回"已接受"（HTTP 202 Accepted）作为响应，然后在后台处理该文件。

使用 BackgroundTasks：基本用法

第一步，导入 BackgroundTasks，然后在你的路径操作函数（path operation function）中声明一个类型为 BackgroundTasks 的参数。

完整示例（对应仓库文件 tutorial001_py310.py）：

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

FastAPI 会为你创建 BackgroundTasks 类型的对象，并作为该参数传入路径操作函数——你不需要自己实例化它。

编写任务函数

任务函数就是要作为背景任务执行的那个函数，它就是一个普通的 Python 函数，可以接收参数：

• 它可以是 async def 异步函数，也可以是普通的 def 同步函数，FastAPI 知道如何处理这两种情况；
• 在上面的示例中，任务函数 write_notification 把内容写入文件（模拟发送邮件）。由于文件写操作没有使用 async 和 await，所以这里用普通的 def 来定义它。

添加背景任务：.add_task()

在路径操作函数内部，通过 .add_task() 方法把任务函数添加到背景任务对象上：

background_tasks.add_task(write_notification, email, message="some notification")

.add_task() 接收的实参依次为：

1. 任务函数本身（write_notification），即要在后台执行的可调用对象；
2. 任意数量的位置参数，按顺序传递给任务函数（示例中的 email）；
3. 任意关键字参数，传递给任务函数（示例中的 message="some notification"）。

从源码看，add_task 的签名与这一说明完全一致。FastAPI 在 fastapi/background.py 中对它做了带类型提示的包装（使用 ParamSpec 保证位置/关键字参数的类型一致性），最终委托给 Starlette 的 BackgroundTasks.add_task 完成实际入队：

def add_task(
    self,
    func: Callable[P, Any],   # 响应发送后调用的函数，def 或 async def 均可
    *args: P.args,
    **kwargs: P.kwargs,
) -> None:
    return super().add_task(func, *args, **kwargs)

与依赖注入（Dependency Injection）结合使用

BackgroundTasks 与 FastAPI 的依赖注入系统完全兼容。你可以在多个层级声明 BackgroundTasks 类型的参数：路径操作函数、依赖（Dependable）、子依赖……FastAPI 知道该怎么做，如何复用同一个对象，使得各处添加的背景任务被合并（merged），然后统一在后台执行。

下面是一个多层级示例（对应仓库文件 tutorial002_an_py310.py）：

from typing import Annotated

from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()


def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)


def get_query(background_tasks: BackgroundTasks, q: str | None = None):
    if q:
        message = f"found query: {q}\n"
        background_tasks.add_task(write_log, message)
    return q


@app.post("/send-notification/{email}")
async def send_notification(
    email: str, background_tasks: BackgroundTasks, q: Annotated[str, Depends(get_query)]
):
    message = f"message to {email}\n"
    background_tasks.add_task(write_log, message)
    return {"message": "Message sent"}

在这个示例中，消息会在响应发送之后写入文件 log.txt：

• 如果请求中携带了 Query 参数 q，依赖 get_query 会添加一个后台任务，把查询内容写入日志；
• 随后，路径操作函数 中创建的另一后台任务，会使用路径参数 email 再写入一条消息。

两条消息最终都出现在 log.txt 中，顺序为 found query: some-query 在前、message to ... 在后。

这一行为有源码层面的明确印证：

• 在 fastapi/dependencies/utils.py 的 solve_dependencies 中，如果 background_tasks 为 None 才新建一个 BackgroundTasks() 对象，否则复用已传入的对象——这正是"多层级共享同一任务集合"的实现基础；递归解决子依赖后，还会把子依赖得到的 background_tasks 回传（background_tasks = solved_result.background_tasks），保证合并结果向上汇总；
• 在 fastapi/routing.py 中，_build_response_args 把 solved_result.background_tasks 设置为响应的 background，即任务集合最终附着在 Response 上，由 ASGI 服务器在响应发送完毕后统一执行；
• 测试用例 test_tutorial002.py 验证了上述合并行为：请求 /send-notification/[e-mail]?q=some-query 后，log.txt 中同时包含依赖中添加的 found query: some-query 和路径操作函数中添加的 message to [e-mail]。

基本用法的端到端验证见 test_tutorial001.py：请求成功后断言 log.txt 中包含 notification for [e-mail]: some notification，证明任务确实在响应返回后执行完毕。

技术细节：来自 Starlette 的 BackgroundTasks

类 BackgroundTasks 直接来自 Starlette 的 starlette.background 模块（在 fastapi/background.py 中可以看到 from starlette.background import BackgroundTasks as StarletteBackgroundTasks，FastAPI 的 BackgroundTasks 是其子类）。

FastAPI 把它直接导入/包含进来，使得你可以从 fastapi 直接导入它，从而避免不小心导入 Starlette 中不带 s 的替代类 BackgroundTask（单数形式是 Starlette 的另一个、面向单一任务的类）。

只使用 BackgroundTasks（而非 BackgroundTask），你就可以把它直接作为路径操作函数的参数使用，让 FastAPI 替你处理其余一切——就像直接使用 Request 对象那样。

当然，你仍然可以在 FastAPI 中单独使用 BackgroundTask（单数），但此时需要自己创建该对象，并返回一个携带它的 Starlette Response（即手动把任务挂到 response.background 上），而不能依赖 FastAPI 的依赖注入自动完成。

注意事项：何时不该只用 BackgroundTasks

当你的后台计算量很大，且不要求必须与请求处理在同一进程中执行（例如不需要共享内存、变量等状态）时，可以考虑使用更大的专用工具，如 Celery。这类工具通常需要更复杂的配置，以及 RabbitMQ 或 Redis 这样的消息/任务队列管理器，但能让你在多个进程、多台服务器上执行后台任务。

反过来，如果你需要访问同一 FastAPI 应用中的变量和对象（共享内存状态），或者只是运行较小的后台任务（比如发送一封邮件通知），直接使用 BackgroundTasks 即可。

小结

• 导入 BackgroundTasks，并在路径操作函数（以及依赖）中声明其类型参数，由 FastAPI 自动注入、跨层级复用同一实例；
• 编写普通 def 或 async def 任务函数，通过 .add_task(函数, *位置参数, **关键字参数) 注册；
• 任务在响应发送给客户端之后执行，适合邮件通知、文件处理等"必须做但不必等"的慢操作；
• 它本质是 Starlette BackgroundTasks 的封装；大规模、跨进程/跨服务器的后台工作则应交给 Celery + 消息队列方案。