FastAPI（41）- Background Task 后台任務

后台任務

• 顧名思義，可以在返回響應后運行后台任務
• 這對於需要在請求后執行特定的操作很有用，且客戶端並不需要在接收響應之前等待該操作完成

 

常見的栗子

• 發送電子郵件通知，由於連接到電子郵件服務器並發送電子郵件往往會比較“緩慢”（幾秒鍾），因此可以立即返回響應並在后台發送電子郵件通知
• 假設您到一個必須經過緩慢處理的文件，可以先返回“已接受”（HTTP 202）響應並在后台處理它

 

實際栗子

創建后台任務要用到的函數

• 創建一個作為后台任務運行的函數，就是一個普通函數
• 可以加 async 也可以不加，FastAPI 將會正確處理它

import time

def write_notification(email: str, message: str = ""):
    # 1、模擬和郵件服務器建立連接
    time.sleep(3)
    with open("text.txt", mode="w") as f:
        # 2、模擬發送郵件
        content = f"message is {message}"
        f.write(content)
        print(content)

  

添加后台任務

#!usr/bin/env python
# -*- coding:utf-8 _*-
"""
# author: 小菠蘿測試筆記
# blog:  https://www.cnblogs.com/poloyy/
# time: 2021/9/29 7:11 下午
# file: 35_background_task.py
"""
import time
import uvicorn
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()


def write_notification(email: str, message: str = ""):
    # 1、模擬和郵件服務器建立連接
    time.sleep(3)
    with open("text.txt", mode="w") as f:
        # 2、模擬發送郵件
        content = f"message is {message}"
        f.write(content)
        print(content)


@app.post("/email/{email}")
async def send_email(
        email: str,
        # 指定參數類型為 BackgroundTasks
        background_task: BackgroundTasks
):
    # 添加后台任務
    # 第一個參數：可調用對象，一般就是函數
    # 后面的參數：write_notification 函數所需要的參數
    background_task.add_task(write_notification, email, message="test_message")
    return {"message": "Notification sent in the background"}


if __name__ == '__main__':
    uvicorn.run(app="35_background_task:app", reload=True, host="127.0.0.1", port=8080)

  

后台任務結合依賴項

#!usr/bin/env python
# -*- coding:utf-8 _*-
"""
# author: 小菠蘿測試筆記
# blog:  https://www.cnblogs.com/poloyy/
# time: 2021/9/29 7:11 下午
# file: 35_background_task.py
"""
import time
from typing import Optional

import uvicorn
from fastapi import FastAPI, BackgroundTasks, Depends

app = FastAPI()


# 后台任務函數
def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)


# 依賴項函數
async def get_query(
        background_task: BackgroundTasks,
        q: Optional[str] = None,
):
    # 如果 q 有值才執行后台任務
    if q:
        message = f"found query: {q}\n"
        background_task.add_task(write_log, message)


@app.post("/email_depends/{email}")
async def send_email(
        email: str,
        background_task: BackgroundTasks,
        q: str = Depends(get_query)
):
    # 執行一次后台任務
    message = f"message to {email}\n"
    background_task.add_task(write_log, message)
    return {"message": "Message sent"}


if __name__ == '__main__':
    uvicorn.run(app="35_background_task:app", reload=True, host="127.0.0.1", port=8080)

• 后台任務可以在任意地方使用，比如路徑操作、依賴項、子依賴項...
• FastAPI 會將所有后台任務合並在一起，然后在后台會按 add_task 的順序運行

 

查看 BackgroundTasks 源碼

• BackgroundTasks 是繼承 BackgroundTask，而 BackgroundTask 是直接來自 starlette.background
• add_task() 第一個參數 func 類型是 Callable，可調用對象，一般傳函數就好啦
• 內部會聲明一個 BackgroundTask 對象，自動調用它的 __call__ 方法
• 可以看到，最終會執行 func()
• func() 函數參數就是 add_task() 函數除第一個參數以外的參數

 

BackgroundTasks 注意事項

• 如果需要執行繁重的后台計算，且可能需要多個進程運行（例如，不需要共享內存、變量等），使用其他更大的工具，如：Celery，效果可能會更好
• 它們往往需要更復雜的配置、消息/作業隊列管理器，如 RabbitMQ 或 Redis，它們允許在多個進程中運行后台任務，尤其是在多個服務器中
• 但是，如果需要從同一個 FastAPI 應用程序訪問變量和對象，或者需要執行小型后台任務（例如發送電子郵件通知），只需使用 BackgroundTasks