FastAPI教程 后台任务

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

这对于需要在请求之后发生的操作很有用，但客户端实际上不必在接收响应之前等待操作完成。

这包括，例如：

• 执行操作后发送的电子邮件通知：由于连接到电子邮件服务器并发送电子邮件往往“缓慢”（几秒钟），因此您可以立即返回响应并在后台发送电子邮件通知。
• 处理数据：例如，假设您收到一个必须经过缓慢处理的文件，您可以返回“已接受”（HTTP 202）响应并在后台处理它。

使用 BackgroundTasks

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

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会知道如何正确处理它。

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

由于写操作不使用asyncand await，我们用 normal 定义函数def：

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()：

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知道在每种情况下要做什么以及如何重用相同的对象，以便将所有后台任务合并在一起，然后在后台运行：

from typing import Optional

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: Optional[str] = None):
    if q:
        message = f"found query: {q}
"
        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}
"
    background_tasks.add_task(write_log, message)
    return {"message": "Message sent"}

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

如果请求中有查询，它将在后台任务中写入日志。

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

技术细节

该类BackgroundTasks直接来自starlette.background.

它直接导入/包含到 FastAPI 中，以便您可以从中导入它fastapi并避免意外导入替代项BackgroundTask（没有s末尾） from starlette.background。

通过只使用BackgroundTasks（而不是BackgroundTask），就可以将其用作路径操作函数参数，并让FastAPI为您处理其余部分，就像Request直接使用对象时一样。

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

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

警告

如果您需要执行繁重的后台计算并且您不一定需要它由同一进程运行（例如，您不需要共享内存、变量等），您可能会受益于使用其他更大的工具，如芹菜。

它们往往需要更复杂的配置、消息/作业队列管理器，如 RabbitMQ 或 Redis，但它们允许您在多个进程中运行后台任务，尤其是在多个服务器中。

要查看示例，请检查Project Generators，它们都包含已配置的 Celery。

但是，如果您需要从同一个FastAPI应用程序访问变量和对象，或者您需要执行小型后台任务（例如发送电子邮件通知），则只需使用BackgroundTasks.

回顾

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