在 AI 服務中,我們經常遇到一些不需要立即回應給用戶的任務,如推理結果的後處理、日誌記錄、或者資料清理等。FastAPI 的 Background Tasks 功能為這類需求提供了優雅的解決方案,讓我們能夠快速回應用戶,同時在背景處理耗時的輔助工作。
在開發 Web 應用時,我們經常遇到一些不想讓使用者在線上等待的耗時操作。最經典的例子就是使用者註冊成功後,系統需要發送一封歡迎郵件。如果將發信的程式直接寫在 API 路由中,使用者就必須等到信件發送完成才能收到「註冊成功」的回應,這會帶來很差的使用者體驗。
為了解決這個問題,FastAPI 提供了一個極度輕量、無需任何額外設定的內建工具:BackgroundTasks
。
BackgroundTasks
是一個透過 FastAPI 的 dependency injection (依賴注入) 系統傳入的工具。它允許你在 API 回應 (Response) 已經回傳給使用者之後,才在背景執行一個或多個函式。
它的核心特性是「即發即棄」(fire-and-forget)。也就是說,你的主程式在提交任務後,就不會再關心它是否成功、何時完成,而是立刻繼續向下執行,最終回傳 HTTP 回應。
要使用它非常簡單,只需要在你的路由函式參數中,宣告一個 BackgroundTasks
型別的變數即可。
from fastapi import FastAPI, BackgroundTasks
import time
app = FastAPI()
def write_log(message: str):
"""一個模擬寫入日誌的慢速函式"""
time.sleep(5) # 模擬 I/O 延遲
with open("log.txt", mode="a") as log_file:
log_file.write(message)
print(f"任務完成: {message.strip()}")
@app.post("/send-notification/{email}")
def send_notification(email: str, background_tasks: BackgroundTasks):
message = f"通知信已發送給 {email}\n"
# 使用 .add_task() 將函式與其參數加入背景任務佇列
background_tasks.add_task(write_log, message)
# 這行會立刻回傳,使用者無需等待 5 秒
return {"message": "正在背景發送通知,請稍候..."}
當你呼叫這個 API 時,會立刻收到 JSON 回應。與此同時,伺服器端的終端機會在 5 秒後印出 "任務完成" 的訊息,並且 log.txt
檔案會被建立與寫入。
此外,在執行 BackgroundTasks
的這 5 秒期間,FastAPI 是可以處理其他 Request 的。
讓我們稍微調整一下程式碼,方便進行展示:
+ import logging
import time
from datetime import datetime
from fastapi import BackgroundTasks, FastAPI
+ # 設定 logger
+ logger = logging.getLogger("uvicorn")
app = FastAPI()
@app.get("/")
def read_root():
+ current_time = datetime.now().strftime("%H:%M:%S")
+ logger.info(f"[{current_time}] 讀取根目錄")
return {"message": "Hello World"}
def write_log(message: str):
"""一個模擬寫入日誌的慢速函式"""
time.sleep(5) # 模擬 I/O 延遲
with open("log.txt", mode="a", encoding="utf-8") as log_file:
log_file.write(message)
+ current_time = datetime.now().strftime("%H:%M:%S")
+ logger.info(f"[{current_time}] 任務完成: {message.strip()}")
@app.post("/send-notification/{email}")
def send_notification(email: str, background_tasks: BackgroundTasks):
message = f"通知信已發送給 {email}\n"
+ current_time = datetime.now().strftime("%H:%M:%S")
+ logger.info(f"[{current_time}] {message.strip()}")
background_tasks.add_task(write_log, message)
return {"message": "正在背景發送通知,請稍候..."}
在發出 /send-notification/{email}
API 請求後的 5 秒內,又打了 3 次 /
API,都可以正常取得回應~
BackgroundTasks
是處理輕量、非關鍵性任務的絕佳選擇,例如記錄日誌、發送不重要的通知等。但如果你的任務至關重要,不容許失敗,或是需要更複雜的管理,那麼我們就需要更專業的工具了。下一篇文章,我們將探討如何處理另一種常見的挑戰:同步阻塞任務。