FastAPI異步任務系列(一) FastAPI后台任務

一、后台任務使用　　

你可以定義后台任務在后台響應之后繼續運行，這對於在請求之后去做一些操作時有用的，但是客戶端不會真正的等待響應中操作的完成。這包括，例如：

• 執行操作后發送電子郵件通知
• 處理數據，比如，后台接收一個文件需要處理，但是可以先給客戶端返回響應，然后后台接着處理

1、使用后台任務

首先，導入BackgroundTask以及在路徑操作函數中定義一個參數，並且聲明類型為BackgroundTasks：

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks): pass

FastAPI將會創建BackgroundTasks類型的對象，並且將其當作參數傳遞。

2、創建一個任務函數

創建一個函數作為后台任務，它是一個可以接收參數的標准函數，可以使用async異步方式或者正常函數方式，FastAPI知道應該如何使用它，在這個例子中，這個后台任務將會寫一個文件：

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_notification(email: str, message=""): with open("log.txt", mode="w") as f: f.write(f"notification for {email}:{message}")


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    pass

3、添加后台任務

在路徑操作函數中，通過".add_task()"函數將任務函數添加到后台任務對象中：

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as f:
        f.write(f"notification for {email}:{message}")


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="notification...") return {"message": "Notification sent in the background"}

.add_task()接收的參數：

• 一個在后台運行的任務函數（write_notification）
• 按照順尋傳遞的一系列參數（email）
• 任何的關鍵字參數（message="notification..."）

二、依賴注入

在依賴注入系統中也可以使用后台任務，你可以聲明一個BackgroundTasks類型的參數，在路徑操作函數、依賴項、子依賴項中等。

from fastapi import BackgroundTasks, FastAPI, Depends
from typing import Optional

app = FastAPI()


def write_log(message: str):
    with open("log.txt", mode="a") as f:
        f.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("/dependency/background_tasks")
async def write_log(q: str = Depends(get_query)):
    if q:
        return {"message": "write log in the background"}