什么是 BackgroundTasks？
BackgroundTasks?是 FastAPI 提供的一個強大工具，它允許你將一些非緊急的、耗時的操作（例如發送郵件、處理數據、調用第三方 API 等）放到“后臺”去執行，而不是讓用戶一直等待這些操作完成。
它的核心思想是：先快速返回響應給客戶端，然后在后臺慢慢處理其他任務。
為什么要使用它？
想象一下這些場景：
用戶注冊：用戶點擊注冊后，你需要在數據庫中創建用戶記錄，然后發送一封歡迎郵件。發送郵件可能需要幾秒鐘（與第三方郵件服務通信）。你不應該讓用戶盯著空白頁面等待郵件發送成功才看到“注冊成功”的提示。更好的體驗是，立即告訴用戶“注冊成功”，然后在后臺默默發送郵件。
文件上傳處理：用戶上傳了一個大視頻，你需要先保存文件，然后啟動一個后臺任務來轉碼、生成縮略圖、分析內容等。這些操作非常耗時，應該立即響應“上傳成功”，處理在后臺進行。
寫入日志或分析數據：記錄用戶的請求行為到數據庫或日志文件，這個操作也不應該阻塞主響應。
在這些情況下，使用?BackgroundTasks?可以顯著提升應用的響應速度和用戶體驗。
工作原理
FastAPI 的?BackgroundTasks?機制非常直觀：
你將一個或多個函數（任務）添加到?
BackgroundTasks?對象中。FastAPI 會先將響應返回給客戶端。
在當前請求的上下文結束后，FastAPI 會在同一個進程中依次執行你添加的所有后臺任務。
重要特點：
同步執行：默認情況下，任務是同步執行的（一個接一個）。如果一個任務很慢，它會阻塞隊列中的下一個任務。
與請求同進程：它適用于輕量級的后臺任務。對于非常耗時或CPU密集型的任務，最好使用更強大的工具（如?
Celery,?RQ,?ARQ），因為它們可以在單獨的進程或 worker 中運行，避免阻塞你的主應用。簡單易用：無需額外的基礎設施（如 Redis、RabbitMQ），開箱即用。
如何使用？
1. 基本使用步驟
a.?導入?BackgroundTasks
b.?在路徑操作函數的參數中聲明它
c.?編寫要后臺運行的函數
d.?使用?add_task()?方法添加任務
e.?返回響應
from fastapi import FastAPI, BackgroundTasks
import timeapp = FastAPI()# 這是一個模擬的耗時任務，例如發送郵件
def send_notification(email: str, message: str):print(f"開始發送郵件給 {email}...")time.sleep(3) # 模擬耗時操作，比如網絡請求print(f"郵件已發送給 {email}：{message}")# 這里可以是實際的發送郵件邏輯，如使用 smtplib@app.post("/register/{username}")
async def register_user(username: str, email: str, background_tasks: BackgroundTasks):# 1. 首先執行主要的快速邏輯（例如將用戶保存到數據庫）# user = create_user_in_db(username, email) <-- 偽代碼print(f"用戶 {username} 已成功創建于數據庫。")# 2. 將耗時任務添加到后臺background_tasks.add_task(send_notification, # 要執行的函數email, # 傳遞給函數的第一個參數f"嗨，{username}！歡迎加入我們！" # 傳遞給函數的第二個參數)# 3. 立即返回響應return {"message": "用戶注冊成功","username": username,"email": email}依賴注入與路徑操作函數
你可以將?BackgroundTasks?不僅注入到路徑操作函數中，還可以注入到依賴項（Dependencies）?中，這使得代碼更加模塊化和可復用。
from fastapi import Dependsdef write_log(message: str):with open("log.txt", mode="a") as log:log.write(message + "\n")def get_query(background_tasks: BackgroundTasks, q: str | None = None):if q:# 在依賴項中添加后臺任務background_tasks.add_task(write_log, f"查詢參數: {q}")return q@app.get("/items/")
async def read_items(query: str = Depends(get_query)):# 這個路徑操作函數本身可能也會添加任務# background_tasks.add_task(...)return {"query": query}
