背景任務¶
您可以定義在返回響應*後*執行的背景任務。
這對於在請求後需要執行的操作很有用，但客戶端在接收響應之前並不需要等待操作完成。
這包含，例如：
- 執行操作後發送的電子郵件通知
- 由於連接到電子郵件伺服器和發送電子郵件往往「很慢」（幾秒鐘），您可以立即返回響應並在背景中發送電子郵件通知。
- 處理數據
- 例如，假設您收到一個必須經過緩慢處理的文件，您可以返回「已接受」（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** 將會知道如何正確處理它。
在這種情況下，任務函數將寫入文件（模擬發送電子郵件）。
由於寫入操作不使用 async 和 await，我們使用普通的 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 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"}
from typing import Annotated, Union
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: Union[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"}
from typing import Union
from fastapi import BackgroundTasks, Depends, FastAPI
from typing_extensions import Annotated
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: Union[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"}
提示
如果可能，請盡量使用 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"}
提示
如果可能，請盡量使用 Annotated 版本。
from typing import Union
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: Union[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，但它們允許您在多個程序中，尤其是在多個伺服器上運行背景任務。
要查看範例，請查看專案產生器，它們都包含已配置好的 Celery。
但是，如果您需要從同一個 FastAPI 應用程式存取變數和物件，或者您需要執行小型背景任務（例如傳送電子郵件通知），則只需使用 BackgroundTasks 即可。
摘要¶
在*路徑操作函式*和依存項目中導入並使用帶有參數的 BackgroundTasks 以新增背景任務。