FastAPI（41）- Background Task 後臺任務

阿新 • • 發佈：2021-09-29

後臺任務

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

常見的栗子

傳送電子郵件通知，由於連線到電子郵件伺服器併發送電子郵件往往會比較“緩慢”（幾秒鐘），因此可以立即返回響應並在後臺傳送電子郵件通知
假設您到一個必須經過緩慢處理的檔案，可以先返回“已接受”（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)