FastAPI教程 后臺(tái)任務(wù)

您可以定義在返回響應(yīng)后運(yùn)行的后臺(tái)任務(wù)。

這對(duì)于需要在請(qǐng)求之后發(fā)生的操作很有用，但客戶端實(shí)際上不必在接收響應(yīng)之前等待操作完成。

這包括，例如：

• 執(zhí)行操作后發(fā)送的電子郵件通知：由于連接到電子郵件服務(wù)器并發(fā)送電子郵件往往“緩慢”（幾秒鐘），因此您可以立即返回響應(yīng)并在后臺(tái)發(fā)送電子郵件通知。
• 處理數(shù)據(jù)：例如，假設(shè)您收到一個(gè)必須經(jīng)過(guò)緩慢處理的文件，您可以返回“已接受”（HTTP 202）響應(yīng)并在后臺(tái)處理它。

使用 BackgroundTasks

首先，BackgroundTasks在路徑操作函數(shù)中導(dǎo)入并定義一個(gè)參數(shù)，類型聲明為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將為您創(chuàng)建類型對(duì)象BackgroundTasks并將其作為該參數(shù)傳遞。

創(chuàng)建任務(wù)函數(shù)

創(chuàng)建一個(gè)作為后臺(tái)任務(wù)運(yùn)行的函數(shù)。

它只是一個(gè)可以接收參數(shù)的標(biāo)準(zhǔn)函數(shù)。

它可以是一個(gè)async def或正常的def函數(shù)，F(xiàn)astAPI會(huì)知道如何正確處理它。

在這種情況下，任務(wù)函數(shù)將寫(xiě)入文件（模擬發(fā)送電子郵件）。

由于寫(xiě)操作不使用asyncand await，我們用 normal 定義函數(shù)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"}

添加后臺(tái)任務(wù)

在您的路徑操作函數(shù)中，使用以下方法將您的任務(wù)函數(shù)傳遞給后臺(tái)任務(wù)對(duì)象.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() 作為參數(shù)接收：

• 要在后臺(tái)運(yùn)行的任務(wù)函數(shù) ( write_notification)。
• 應(yīng)該按順序傳遞給任務(wù)函數(shù)的任何參數(shù)序列 ( email)。
• 應(yīng)該傳遞給任務(wù)函數(shù)的任何關(guān)鍵字參數(shù) ( message="some notification")。

依賴注入

使用BackgroundTasks也適用于依賴注入系統(tǒng)，您可以BackgroundTasks在多個(gè)級(jí)別聲明類型的參數(shù)：在路徑操作函數(shù)中，在依賴項(xiàng)（可靠）中，在子依賴項(xiàng)中等。

FastAPI知道在每種情況下要做什么以及如何重用相同的對(duì)象，以便將所有后臺(tái)任務(wù)合并在一起，然后在后臺(tái)運(yùn)行：

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

在此示例中，消息將在發(fā)送響應(yīng)后寫(xiě)入log.txt文件。

如果請(qǐng)求中有查詢，它將在后臺(tái)任務(wù)中寫(xiě)入日志。

然后在路徑操作函數(shù)中生成的另一個(gè)后臺(tái)任務(wù)將使用email路徑參數(shù)寫(xiě)入一條消息。

技術(shù)細(xì)節(jié)

該類BackgroundTasks直接來(lái)自starlette.background.

它直接導(dǎo)入/包含到 FastAPI 中，以便您可以從中導(dǎo)入它fastapi并避免意外導(dǎo)入替代項(xiàng)BackgroundTask（沒(méi)有s末尾） from starlette.background。

通過(guò)只使用BackgroundTasks（而不是BackgroundTask），就可以將其用作路徑操作函數(shù)參數(shù)，并讓FastAPI為您處理其余部分，就像Request直接使用對(duì)象時(shí)一樣。

它仍然可以BackgroundTask在 FastAPI 中單獨(dú)使用，但您必須在代碼中創(chuàng)建對(duì)象并返回Response包含它的 Starlette 。

您可以在Starlette 的后臺(tái)任務(wù)官方文檔中查看更多詳細(xì)信息。

警告

如果您需要執(zhí)行繁重的后臺(tái)計(jì)算并且您不一定需要它由同一進(jìn)程運(yùn)行（例如，您不需要共享內(nèi)存、變量等），您可能會(huì)受益于使用其他更大的工具，如芹菜。

它們往往需要更復(fù)雜的配置、消息/作業(yè)隊(duì)列管理器，如 RabbitMQ 或 Redis，但它們?cè)试S您在多個(gè)進(jìn)程中運(yùn)行后臺(tái)任務(wù)，尤其是在多個(gè)服務(wù)器中。

要查看示例，請(qǐng)檢查Project Generators，它們都包含已配置的 Celery。

但是，如果您需要從同一個(gè)FastAPI應(yīng)用程序訪問(wèn)變量和對(duì)象，或者您需要執(zhí)行小型后臺(tái)任務(wù)（例如發(fā)送電子郵件通知），則只需使用BackgroundTasks.

回顧

BackgroundTasks在路徑操作函數(shù)和依賴項(xiàng)中導(dǎo)入并使用參數(shù)添加后臺(tái)任務(wù)。