后台任务 BackgroundTasks

有些活儿，用户其实不需要等它做完。比如用户注册成功，你只想立刻告诉他”注册好了”，至于发欢迎邮件这种可能要花好几秒的事，完全可以等响应返回后再慢慢做。FastAPI 的 BackgroundTasks 就是干这个的。

40-1 后台任务是什么

后台任务指的是：在接口已经把响应返回给客户端之后，才在背后执行的任务。客户端不等它，体验更流畅。

常见场景有三类：

• 发邮件通知。连邮件服务器往往要几秒，没必要让用户干等。
• 写日志或做统计。请求处理完了，顺手记一笔。
• 处理上传的文件。先回个”已接收”，再慢慢转码或分析。

关键点：这些任务的执行时间点，是在响应发出之后。用户早拿到结果了，后台还在忙。

这种设计还有个附带好处：接口能立刻回复，不会因为后台慢活而拖垮响应速度。对于”接收了但还没处理完”的场景，业界常返回 HTTP 202（Accepted）状态码，明确告诉客户端”我收到了，正在排期处理”。你只要在路径操作里写上 status_code=202 即可，后台任务照常登记。

40-2 声明 BackgroundTasks 参数

用法很简单。先从 fastapi 导入 BackgroundTasks，然后在路径操作函数里声明一个同类型的参数。FastAPI 会自动帮你创建好这个对象并传进来。

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

只要函数签名里出现了 background_tasks: BackgroundTasks 这种声明，FastAPI 就懂了：这是要往后台塞任务。

40-3 写一个任务函数

任务函数就是普通的 Python 函数，能收参数。它可以是 async def，也可以是普通 def，FastAPI 都能正确处理。

下面这个任务模拟”发邮件”——实际上我们往文件里写一行内容，方便你直接跑起来看效果。因为它用的是普通文件写操作（不支持 await），所以写成 def：

def write_notification(email: str, message: str = ""):
    with open("log.txt", "w", encoding="utf-8") as f:
        content = f"notification for {email}: {message}\n"
        f.write(content)

注意它有两个参数：email 和 message。等会儿我们调用任务时，会把这些参数传进去。

40-4 用 add_task 添加任务

在路径操作函数内部，调用 background_tasks.add_task(...) 来登记一个后台任务。add_task 的第一个参数是任务函数，后面跟着要传给它的位置参数和关键字参数。

@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"}

这个接口的逻辑是：先把任务登记进 background_tasks，然后立刻返回 {"message": ...}。客户端几乎瞬间收到响应。等响应发完，FastAPI 才去执行 write_notification，把内容写进 log.txt。

启动后访问 /send-notification/alice@example.com，你会马上看到返回，随后 log.txt 里出现一行通知记录。

40-5 任务函数也可以是异步的

如果后台任务里要调异步库（比如异步发邮件），把任务函数写成 async def 即可，其余写法不变：

import asyncio


async def send_email_async(email: str):
    # 假设这里是异步邮件客户端
    await asyncio.sleep(1)
    print(f"email sent to {email}")


@app.post("/notify/{email}")
async def notify(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email_async, email)
    return {"message": "queued"}

FastAPI 会判断任务函数是不是 async def，然后选对的方式去跑它。你不用操心。

40-6 和依赖注入结合

BackgroundTasks 还能和依赖注入搭配，这是个很实用的组合。你在依赖里也可以声明 BackgroundTasks 参数，往里加任务。FastAPI 聪明地把同一个请求里登记的任务合并到一起，统一在响应后执行。

下面这个例子：依赖函数 get_query 检查有没有查询参数 q，有就登记一条写日志任务；路径操作本身再登记一条。两条都会执行。

from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()


def write_log(message: str):
    with open("log.txt", "a", encoding="utf-8") as f:
        f.write(message)


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


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

访问 /send-notification/bob?q=hello，响应返回后，log.txt 里会多两行：一行来自依赖登记的查询日志，一行来自路径操作登记的信息日志。

40-7 两个容易踩的小坑

第一，BackgroundTasks 和 BackgroundTask 不是一个东西。请务必导入复数形式的 BackgroundTasks（带 s），它是 FastAPI 直接从 Starlette 包好、方便你当参数用的版本。单数 BackgroundTask 得你自己创建对象再塞进 Response 里，麻烦得多。

第二，后台任务跑在同一个进程里。如果你的活儿特别重（比如要转码一小时的视频），而且不关心和当前程序共享内存，那就该上更专业的工具，比如 Celery，配 RabbitMQ 或 Redis 做任务队列，还能跨多台服务器。但只是发个邮件、记笔日志这种轻活，BackgroundTasks 足够好用。

40-8 小结

这一章你学会了后台任务：

• 用 BackgroundTasks 类型声明参数，FastAPI 自动注入对象。
• 用 background_tasks.add_task(函数, 参数...) 登记任务。
• 任务函数可以是 async def 或 def，且能在依赖里登记，自动合并。
• 轻量收尾活用它，超重的活交给 Celery 之类的专业队列。

下一章我们讲中间件（Middleware）和跨域（CORS），它们都是在”每个请求进出时”统一动手脚的机制。