后台任务¶

你可以定义在返回响应之后运行的后台任务。

这对于那些需要在请求处理完毕后执行，但客户端无需等待其完成即可接收响应的操作非常有用。

例如，这包括：

执行操作后发送的电子邮件通知
- 由于连接邮件服务器并发送邮件通常比较“缓慢”（需要几秒钟），你可以立即返回响应，并在后台发送邮件通知。
数据处理
- 例如，假设你接收到一个必须经过缓慢处理过程的文件，你可以返回“已接收”（HTTP 202）响应，并在后台处理该文件。

使用 `BackgroundTasks`¶

首先，导入 BackgroundTasks，并在你的路径操作函数中定义一个类型声明为 BackgroundTasks 的参数。

Python 3.10+

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 类型的对象，并将其作为该参数传递。

创建任务函数¶

创建一个要在后台运行的任务函数。

它只是一个可以接收参数的标准函数。

它可以是 async def 函数或普通 def 函数，FastAPI 知道如何正确处理它。

在这种情况下，任务函数将写入一个文件（模拟发送电子邮件）。

由于写入操作不使用 async 和 await，我们使用普通的 def 定义该函数。

Python 3.10+

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

添加后台任务¶

在路径操作函数内部，使用 .add_task() 方法将你的任务函数传递给后台任务对象。

Python 3.10+

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

.add_task() 接收的参数包括：

要在后台运行的任务函数（write_notification）。
按顺序传递给任务函数的任何参数序列（email）。
传递给任务函数的任何关键字参数（message="some notification"）。

依赖注入¶

使用 BackgroundTasks 同样适用于依赖注入系统，你可以在多个层级声明 BackgroundTasks 类型的参数：在路径操作函数中、在依赖项（可依赖对象）中、在子依赖项中等等。

FastAPI 知道在每种情况下该怎么做以及如何重用同一个对象，因此所有后台任务会被合并，并在响应发送后在后台统一运行。

Python 3.10+

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

🤓 其他版本和变体

Python 3.10+ - 非 Annotated

提示

如果可能，请优先使用 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: str = Depends(get_query)
):
    message = f"message to {email}\n"
    background_tasks.add_task(write_log, message)
    return {"message": "Message sent"}

在此示例中，消息将在响应发送之后写入 log.txt 文件。

如果请求中包含查询参数，它将会在后台任务中被写入日志。

然后，在路径操作函数中生成的另一个后台任务将使用 email 路径参数写入一条消息。

技术细节¶

BackgroundTasks 类直接来自于 starlette.background。

它被直接导入/包含进 FastAPI 中，以便你可以从 fastapi 进行导入，从而避免意外地从 starlette.background 导入另一个 BackgroundTask（结尾没有 s）。

仅使用 BackgroundTasks（而不是 BackgroundTask），就可以将其用作路径操作函数参数，并让 FastAPI 为你处理剩下的工作，就像直接使用 Request 对象一样。

虽然在 FastAPI 中仍然可以单独使用 BackgroundTask，但你必须在代码中手动创建该对象，并返回包含该对象的 Starlette Response。

你可以在 Starlette 关于后台任务的官方文档中查看更多详细信息。

注意事项¶

如果你需要执行繁重的后台计算，并且不一定非要由同一个进程运行（例如，你不需要共享内存、变量等），那么你可能更适合使用像 Celery 这样的大型工具。

它们通常需要更复杂的配置和消息/任务队列管理器（如 RabbitMQ 或 Redis），但它们允许你在多个进程，甚至是多个服务器上运行后台任务。

但如果你只需要访问同一个 FastAPI 应用中的变量和对象，或者只是执行一些小的后台任务（如发送电子邮件通知），则可以直接使用 BackgroundTasks。

总结¶

导入并使用带有参数的 BackgroundTasks，可在路径操作函数和依赖项中添加后台任务。

后台任务¶

使用 BackgroundTasks¶

创建任务函数¶

添加后台任务¶

依赖注入¶

技术细节¶

注意事项¶

总结¶

使用 `BackgroundTasks`¶