FastAPI 進階知識(四) 后台任務

作者：麥克煎蛋   出處：https://www.cnblogs.com/mazhiyong/ 轉載請保留這段聲明，謝謝！

 

有時候我們需要在request執行之后繼續一些操作，但終端並不需要等待這些操作完成才能收到response。

其中一些場景舉例如下：

1、在執行完request之后發送郵件通知。

2、收到文件之后對文件進行二次處理。

這些操作都需要一定的處理時間，但與返回給終端的response並無直接關系。

我們可以通過定義后台任務BackgroundTasks，來實現這個功能。

 

使用 BackgroundTasks

首先我們需要導入BackgroundTasks，並在路徑操作函數中添加BackgroundTasks類型的參數。

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


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

任務函數

然后我們需要創建一個在后台任務中實際執行的函數：任務函數。

這是一個標准函數。這個函數可以是async def或者普通def的函數。

這里我們創建一個把指定內容寫入文件的函數。

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)

添加后台任務

最后我們需要把任務函數添加到后台任務中去。

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參數。

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