FastAPI 課程 – 背景任務（Background Tasks）與依賴注入結合

簡介

在 Web API 開發中，我們常會遇到「需要非同步、但不影響使用者回應的工作」── 例如寄送驗證信、產生報表、寫入日誌或執行長時間的計算。若直接在路由函式內執行這類工作，使用者必須等待所有任務完成才會收到回應，會嚴重拖慢 API 的效能與使用體驗。

FastAPI 提供了 BackgroundTasks 這個輕量級機制，讓開發者可以把這類「背景」工作延後執行，同時保持路由的同步或非同步介面不變。而 依賴注入（Dependency Injection, DI） 是 FastAPI 的核心特性之一，讓我們可以在路由、背景任務、甚至測試環境中輕鬆注入資料庫連線、設定檔或其他服務。

本篇文章將說明 如何把 BackgroundTasks 與依賴注入結合，讓背景工作在取得所需資源後安全、有效地執行。文章以 繁體中文（台灣） 撰寫，適合剛入門的開發者，也能為中階開發者提供實務參考。

---

核心概念

1. BackgroundTasks 基礎

BackgroundTasks 是 FastAPI 內建的類別，實際上是 Starlette 的 BackgroundTask 包裝。使用方式非常簡潔：

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def write_log(message: str):
    """簡單的寫檔範例，模擬寫入日誌"""
    with open("log.txt", "a") as f:
        f.write(message + "\n")

@app.post("/items/")
async def create_item(name: str, background_tasks: BackgroundTasks):
    # 立刻回傳成功訊息
    background_tasks.add_task(write_log, f"Item created: {name}")
    return {"msg": f"Item {name} is being processed"}

• background_tasks.add_task(func, *args, **kwargs)：將 func 加入背景佇列，FastAPI 會在回傳 HTTP 回應後自行呼叫它。
• 執行環境：FastAPI 會在同一個事件迴圈（event loop）內以 非阻塞 方式執行任務，若使用同步函式，仍會在背景執行，但會佔用執行緒。

⚠️ 注意：BackgroundTasks 只適合 短時間、非 CPU 密集 的工作；若需要長時間或高 CPU 負載的任務，建議使用 Celery、RQ 或其他訊息佇列系統。

---

2. 依賴注入（Dependency Injection）概念

FastAPI 透過 Depends 讓開發者在路由、背景任務、甚至在其他依賴中注入資源。典型的例子是注入資料庫 Session：

from fastapi import Depends

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

在路由中：

@app.get("/users/")
def read_users(db: Session = Depends(get_db)):
    return db.query(User).all()

依賴函式可以是 同步或非同步，也可以返回 生成器（yield） 以支援「前置」與「後置」清理動作。

---

3. 為什麼要把 BackgroundTasks 與 DI 結合？

單純的 add_task(write_log, ...) 雖然簡潔，但 無法直接取得依賴注入的資源（例如資料庫連線、設定檔、第三方服務客戶端）。如果背景任務需要這些資源，就必須自行在函式內建立或傳遞，這會導致：

1. 程式碼重複：每個背景任務都要自行建立相同的資源。
2. 資源管理不一致：無法保證背景任務的「前置」與「後置」清理（例如關閉 DB 連線）。
3. 測試困難：背景任務的依賴無法被 mock。

透過 將依賴注入的結果作為 add_task 的參數，或 使用 Depends 產生一個「可呼叫」的背景任務，我們可以讓背景工作自動取得所需資源，保持程式碼乾淨、易測試且資源管理一致。

---

程式碼範例

以下示範 5 個實用範例，從最簡單的背景任務到結合依賴注入、資料庫、設定與自訂類別的完整流程。

範例 1️⃣ 基本背景任務 + 直接傳遞參數

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_email(to: str, subject: str, body: str):
    """模擬寄送 Email（實務上會使用 async SMTP 客戶端）"""
    print(f"Sending email to {to}: {subject}")

@app.post("/register/")
async def register_user(email: str, background_tasks: BackgroundTasks):
    # 立刻回應使用者，背景任務負責寄送驗證信
    background_tasks.add_task(send_email, email, "Welcome!", "Thank you for registering.")
    return {"msg": "Registration successful, please check your email."}

重點：此範例說明 如何把任務加入佇列，但尚未使用 DI。

---

範例 2️⃣ 使用 DI 取得設定檔（Config）並傳入背景任務

from fastapi import FastAPI, BackgroundTasks, Depends
from pydantic import BaseSettings

class Settings(BaseSettings):
    smtp_server: str = "smtp.example.com"
    smtp_port: int = 587

def get_settings() -> Settings:
    return Settings()   # 在實務上會使用 Singleton 或環境變數

def send_email_with_config(to: str, subject: str, body: str, cfg: Settings):
    # 這裡示意使用設定檔的資訊
    print(f"Connecting to {cfg.smtp_server}:{cfg.smtp_port} to send email to {to}")

app = FastAPI()

@app.post("/invite/")
async def invite_user(email: str,
                     background_tasks: BackgroundTasks,
                     cfg: Settings = Depends(get_settings)):
    # 把 DI 取得的設定直接傳給背景任務
    background_tasks.add_task(send_email_with_config, email,
                              "You are invited!", "Please join us.", cfg)
    return {"msg": "Invitation sent in background"}

技巧：直接把 cfg（依賴）作為 add_task 的參數，FastAPI 會在回應前先解析依賴，確保背景任務得到正確的設定物件。

---

範例 3️⃣ 結合資料庫 Session（SQLAlchemy）與背景任務

from fastapi import FastAPI, BackgroundTasks, Depends
from sqlalchemy.orm import Session
from .database import SessionLocal, engine, Base, User

Base.metadata.create_all(bind=engine)

def get_db() -> Session:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def log_user_creation(user_id: int, db: Session):
    """在背景任務中寫入 audit log"""
    db.execute(
        "INSERT INTO audit_log (user_id, action) VALUES (:uid, :act)",
        {"uid": user_id, "act": "created"},
    )
    db.commit()

app = FastAPI()

@app.post("/users/")
async def create_user(name: str,
                      background_tasks: BackgroundTasks,
                      db: Session = Depends(get_db)):
    # 先同步寫入使用者資料
    new_user = User(name=name)
    db.add(new_user)
    db.commit()
    db.refresh(new_user)

    # 再把背景任務加入佇列，傳入同一個 db session
    # 注意：此處 **不能直接傳遞同一個 Session** 給背景任務，因為
    #       回應結束後依賴會關閉 Session。解法是 **在背景任務內部重新取得 Session**。
    background_tasks.add_task(log_user_creation, new_user.id, SessionLocal())
    return {"id": new_user.id, "name": new_user.name}

說明：

• 不要把已關閉的 Session 傳給背景任務。在範例中，我們使用 SessionLocal() 重新產生一個獨立的 Session，確保背景任務在執行時仍能存取資料庫。
• 若你希望背景任務共用同一個依賴的「建立」與「關閉」流程，可改寫成 依賴生成器（見範例 4）。

---

範例 4️⃣ 使用 依賴生成器 包裝背景任務（最乾淨的寫法）

from fastapi import FastAPI, BackgroundTasks, Depends
from sqlalchemy.orm import Session
from .database import SessionLocal

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def background_logger(db: Session = Depends(get_db)):
    """回傳一個可直接呼叫的函式，已經注入好 db"""
    def logger(user_id: int):
        db.execute(
            "INSERT INTO audit_log (user_id, action) VALUES (:uid, :act)",
            {"uid": user_id, "act": "created"},
        )
        db.commit()
    return logger

app = FastAPI()

@app.post("/users2/")
async def create_user2(name: str,
                       background_tasks: BackgroundTasks,
                       logger = Depends(background_logger)):
    # 同步建立使用者
    db = next(get_db())   # 直接取一次 db 用於建立
    new_user = User(name=name)
    db.add(new_user)
    db.commit()
    db.refresh(new_user)

    # 背景任務只需要傳入 user_id，logger 已經封裝好 db
    background_tasks.add_task(logger, new_user.id)
    return {"id": new_user.id, "name": new_user.name}

優點：

• 依賴注入的生命週期 被完整保留：background_logger 內部使用 Depends(get_db)，FastAPI 會在背景任務執行前建立 Session，任務結束後自動關閉。
• 程式碼更簡潔：路由只需要 logger 這個「可呼叫」的函式，不必自行管理 DB。

---

範例 5️⃣ 結合自訂服務類別（EmailClient）與 BackgroundTasks

from fastapi import FastAPI, BackgroundTasks, Depends
import httpx

class EmailClient:
    """簡易的非同步 Email 客戶端，實務上會包裝外部 API"""
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.mailservice.com/v1"

    async def send(self, to: str, subject: str, body: str):
        async with httpx.AsyncClient() as client:
            await client.post(
                f"{self.base_url}/send",
                json={"to": to, "subject": subject, "body": body},
                headers={"Authorization": f"Bearer {self.api_key}"},
                timeout=10,
            )

def get_email_client() -> EmailClient:
    # 假設 API_KEY 從環境變數或設定檔讀取
    return EmailClient(api_key="YOUR_API_KEY")

app = FastAPI()

@app.post("/newsletter/")
async def send_newsletter(
    email: str,
    background_tasks: BackgroundTasks,
    client: EmailClient = Depends(get_email_client),
):
    # 把非同步的 send 方法包裝成同步呼叫，讓 BackgroundTasks 能直接使用
    async def _send():
        await client.send(email, "Monthly Newsletter", "Here is our news...")

    # BackgroundTasks 只能接受同步函式，若要執行 async 必須自行建立執行緒或使用 asyncio.create_task
    # 這裡示範使用 asyncio.create_task（FastAPI 會在回應後自動執行）
    import asyncio
    background_tasks.add_task(asyncio.create_task, _send())
    return {"msg": "Newsletter will be sent shortly"}

關鍵點：

• EmailClient 為自訂服務類別，透過 DI 取得實例。
• 若背景任務是 非同步，可使用 asyncio.create_task 包裝，或改用 anyio 的 run_in_threadpool。
• 這種寫法讓 測試時只要 mock EmailClient，即可驗證背景任務是否正確被呼叫。

---

常見陷阱與最佳實踐

陷阱	為什麼會發生	解決方案
背景任務使用已關閉的依賴（如 DB Session）	Depends 產生的資源在路由結束後會被自動關閉	1️⃣ 於背景任務內重新取得資源；2️⃣ 使用 依賴生成器 包裝背景任務（範例 4）
背景任務阻塞事件迴圈	使用大量同步 I/O（例如 time.sleep、同步 HTTP 請求）	改為 非同步（async def + await）或使用 run_in_threadpool
背景任務例外未被捕捉	FastAPI 只會在任務結束時印出錯誤，且不會回傳給使用者	在背景函式內 加入 try/except，必要時使用日誌系統（如 loguru）記錄
任務過多導致記憶體激增	每個請求都加入大量背景任務，且未限制佇列長度	使用 訊息佇列（Celery、RQ） 處理長時間或大量任務；或在應用層加入 佇列上限
依賴注入的參數順序錯誤	BackgroundTasks 必須放在路由參數的最後（或使用 * 位置參數）	確保 background_tasks: BackgroundTasks 在最後，或使用 **kwargs 收集其他參數

最佳實踐

1. 保持背景任務短小：盡量只做 I/O（寄信、寫檔、呼叫外部 API），避免 CPU 密集計算。
2. 使用依賴生成器包裝背景任務：如範例 4，可讓資源的生命週期自動管理。
3. 加入錯誤處理與日誌：背景任務失敗不會直接影響使用者，但應該被記錄以便排查。
4. 測試時 mock 依賴：使用 TestClient + override_dependency，把背景任務換成 no‑op 或 assert 呼叫。
5. 評估是否需要外部佇列：若業務需求是「大量、長時間」的工作，請改用 Celery、RQ、或 AWS SQS 等方案。

---

實際應用場景

場景	為何適合使用 BackgroundTasks + DI
使用者註冊後寄送驗證信	只需簡單的 SMTP 客戶端（DI 注入設定），即時回應使用者，背景任務負責寄信。
上傳檔案後產生縮圖	圖片處理需要 Pillow 或 opencv，可把設定（尺寸、品質）注入背景任務；若處理時間較長，仍建議使用外部佇列。
訂單建立後寫入審計日志	使用資料庫 Session（DI）寫入 audit_log，背景任務確保不阻塞訂單 API。
定時發送報表	雖然是「定時」工作，但可在 API 中觸發「立即」產生報表的背景任務，注入報表產生服務（DI）。
呼叫第三方支付平台的非同步回呼	背景任務負責向支付平台發送確認訊息，DI 注入 httpx.AsyncClient 或自訂的 PaymentClient。

---

總結

• BackgroundTasks 為 FastAPI 提供了 輕量級的背景執行機制，適合處理短暫、非阻塞的 I/O 工作。
• 依賴注入 讓我們可以在背景任務中 安全、統一地取得資源（設定、資料庫、外部服務客戶端），避免重複程式碼與資源管理錯誤。
• 透過 依賴生成器 包裝背景任務（範例 4），可以讓背景任務的生命週期與路由保持一致，自動管理資源的建立與關閉。
• 在實務開發中，先評估任務的性質：若是短時間 I/O，直接使用 BackgroundTasks；若是長時間或高 CPU 負載，請改用 Celery、RQ 等訊息佇列。
• 錯誤處理、日誌、測試 是不可忽視的部分，確保背景任務失敗不會悄悄消失，且能在 CI/CD 中得到驗證。

掌握 BackgroundTasks 與依賴注入的結合，你就能寫出 高效、可維護且易測試 的 FastAPI 應用，讓 API 在提供即時回應的同時，仍能完成各種後端工作。祝開發順利，快把這些技巧應用到你的下個專案吧！ 🚀