FastAPI：依賴注入系統（Dependency Injection）─ Background Tasks + Dependencies

---

簡介

在 FastAPI 中，依賴注入（Dependency Injection, DI）是讓路由函式保持乾淨、可測試與可重用的核心機制。除了傳統的參數驗證、資料庫連線等常見需求，FastAPI 也提供了 BackgroundTask 物件，讓我們能在回應送出之後，非同步地執行長時間工作（例如發送電子郵件、寫入日誌、觸發第三方 API）。

把 Background Tasks 與 Dependencies 結合，意味著：

• 任務可以自動取得所需資源（DB 連線、設定、認證資訊等）而不必在每個路由裡重複寫程式碼。
• 背景工作不會阻塞主請求，提升 API 的回應速度與使用者體驗。
• 測試更容易：只要替換或 mock 依賴，即可驗證背景任務是否正確被排程。

本篇文章將從概念說明、實作範例、常見陷阱與最佳實踐，逐步帶你掌握「background tasks + dependencies」的使用方式，並提供實務情境的應用示例。

---

核心概念

1. 什麼是 BackgroundTask？

BackgroundTask 是 FastAPI 內建的類別，屬於 Starlette 的 BackgroundTask。當你在路由函式中接受一個 BackgroundTasks 參數，FastAPI 會在回傳 HTTP 回應之後，自動呼叫 background_tasks.add_task(func, *args, **kwargs) 所排入的工作。

重點：背景工作仍在同一個 Python 進程中執行，若需要真正的併發或分散式處理，請考慮 Celery、RQ 等工具。

2. 依賴注入（Dependency Injection）

FastAPI 允許你定義 依賴函式（dependency function），透過 Depends 把它注入到路由或其他依賴中。依賴函式可以：

• 建立/釋放資源（例如 DB 連線、Redis 客戶端）
• 讀取設定檔或環境變數
• 執行認證與授權檢查

結合 BackgroundTasks，我們可以在背景任務裡直接使用這些資源，而不必在每個任務內手動建立。

3. 為什麼要把兩者結合？

• 解耦：背景任務不需要知道「誰」呼叫它，只負責執行業務邏輯。
• 資源共享：同一個請求的依賴（例如同一個 DB session）可以在背景任務中重複使用，避免重複連線。
• 測試友善：測試時只要提供一個「假」依賴，就能驗證背景任務的行為，而不必真的發送郵件或寫檔。

---

程式碼範例

以下示範 4 個實用範例，涵蓋最常見的使用情境。程式碼均以 Python 為例，使用 FastAPI 2.x 版。

3.1 基礎範例：在回應後寄送驗證信

from fastapi import FastAPI, BackgroundTasks, Depends
from pydantic import BaseModel, EmailStr
import smtplib

app = FastAPI()

class UserCreate(BaseModel):
    email: EmailStr
    password: str

def send_verification_email(to_email: str):
    """模擬寄送驗證信的函式（實務上應該使用 async email client）"""
    with smtplib.SMTP("localhost") as client:
        client.sendmail(
            from_addr="no-reply@example.com",
            to_addrs=[to_email],
            msg=f"Subject: Verify your account\n\nPlease verify your email."
        )
    print(f"Verification email sent to {to_email}")

@app.post("/users/")
async def create_user(
    user: UserCreate,
    background_tasks: BackgroundTasks,
):
    # 這裡通常會寫入資料庫、加密密碼等
    # ...

    # 把寄信任務加入背景工作
    background_tasks.add_task(send_verification_email, user.email)
    return {"msg": "User created, verification email will be sent."}

說明

• background_tasks 直接作為路由參數注入。
• send_verification_email 是普通同步函式，FastAPI 會在回應送出後執行它。

---

3.2 使用依賴取得資料庫 Session，並在背景任務寫入日誌

from fastapi import FastAPI, Depends, BackgroundTasks
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker, Session

DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

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

def write_audit_log(db: Session, user_id: int, action: str):
    """把審計紀錄寫入資料庫"""
    db.execute(
        text("INSERT INTO audit_log (user_id, action) VALUES (:uid, :act)"),
        {"uid": user_id, "act": action},
    )
    db.commit()
    print(f"Audit log saved for user {user_id}")

app = FastAPI()

@app.post("/items/{user_id}")
async def create_item(
    user_id: int,
    background_tasks: BackgroundTasks,
    db: Session = Depends(get_db),
):
    # 假設這裡有寫入 item 的邏輯
    # ...

    # 把審計寫入任務加入背景工作，直接使用同一個 db session
    background_tasks.add_task(write_audit_log, db, user_id, "create_item")
    return {"msg": "Item created, audit will be logged in background"}

說明

• get_db 是典型的 依賴函式，回傳一個 SQLAlchemy Session。
• 背景任務直接接收 db 參數，避免在任務內再次建立連線。
• 注意：若背景任務執行時間較長，務必確保 db 仍然有效（在本例中 db 於請求結束前不會關閉）。

---

3.3 結合 async 背景任務與依賴：發送 Slack 訊息

import httpx
from fastapi import FastAPI, BackgroundTasks, Depends

app = FastAPI()

SLACK_WEBHOOK = "https://hooks.slack.com/services/XXXXX/XXXXX/XXXXX"

def get_slack_client() -> httpx.AsyncClient:
    """返回一個可重用的 async http client"""
    client = httpx.AsyncClient()
    try:
        yield client
    finally:
        # 在應用關閉時關閉 client
        client.aclose()

async def post_to_slack(client: httpx.AsyncClient, message: str):
    payload = {"text": message}
    await client.post(SLACK_WEBHOOK, json=payload)
    print("Slack message posted")

@app.post("/orders/{order_id}/notify")
async def notify_order(
    order_id: int,
    background_tasks: BackgroundTasks,
    client: httpx.AsyncClient = Depends(get_slack_client),
):
    # 這裡可能會先處理訂單...
    # ...

    # 把 async 任務加入背景工作
    background_tasks.add_task(post_to_slack, client, f"Order {order_id} created")
    return {"msg": f"Order {order_id} received, notification will be sent"}

說明

• httpx.AsyncClient 可被視為 依賴，在整個應用生命週期內共用。
• post_to_slack 為 async 函式，FastAPI 會自動在背景執行緒中跑 await。

---

3.4 多層依賴：先驗證使用者 → 取得專屬設定 → 背景任務寫檔

from fastapi import FastAPI, Depends, HTTPException, BackgroundTasks, Header
from pathlib import Path

app = FastAPI()

# ---------- 第 1 層：認證 ----------
def get_current_user(x_token: str = Header(...)):
    if x_token != "secret-token":
        raise HTTPException(status_code=401, detail="Invalid token")
    return {"username": "alice"}

# ---------- 第 2 層：根據使用者取得設定 ----------
def get_user_config(user: dict = Depends(get_current_user)):
    # 假設每位使用者都有自己的 config 檔案
    config_path = Path(f"./configs/{user['username']}.json")
    if not config_path.exists():
        raise HTTPException(status_code=404, detail="Config not found")
    return {"config_path": config_path}

# ---------- 背景任務 ----------
def write_report(config_path: Path, report_data: str):
    report_file = config_path.parent / "reports.txt"
    with open(report_file, "a", encoding="utf-8") as f:
        f.write(report_data + "\n")
    print(f"Report written to {report_file}")

@app.post("/report/")
async def generate_report(
    data: str,
    background_tasks: BackgroundTasks,
    cfg: dict = Depends(get_user_config),
):
    # 立即回應使用者
    background_tasks.add_task(write_report, cfg["config_path"], data)
    return {"msg": "Report generation scheduled"}

說明

• get_current_user 與 get_user_config 為 多層依賴，最終產生的 config_path 被傳入背景任務。
• 這種寫法非常適合 租戶化（multi‑tenant） 系統，讓每個使用者的背景任務自動使用自己的資源。

---

常見陷阱與最佳實踐

陷阱	可能的後果	解決方案或最佳做法
在背景任務中直接使用 await 的同步函式	會拋出 RuntimeError: cannot schedule new futures	確保背景任務是 同步（使用 add_task) 或 async（使用 add_task 且函式本身為 async）
依賴產生的資源在請求結束後被關閉（如 DB session）	背景任務執行時出現 ProgrammingError: Closed connection	在依賴中使用 yield，讓 FastAPI 在請求結束前保持資源；或在背景任務內重新取得資源
背景任務執行時間過長	會佔用同一個工作執行緒，導致其他背景任務被阻塞	若任務可能超過數秒，考慮 將工作外移至訊息佇列（Celery、RQ）
忘記在測試中 mock 背景任務	測試會真的發送郵件、寫檔，造成副作用	使用 unittest.mock 或 pytest 的 monkeypatch 把 background_tasks.add_task 替換成 dummy 函式
在背景任務裡直接使用 request 物件	request 只在原始請求生命週期內有效，背景任務執行時已失效	把需要的資訊（如使用者 ID、header）提前抽取，作為參數傳入背景任務

其他最佳實踐

1. 明確命名背景任務：send_welcome_email、log_audit 等，使程式碼易讀。
2. 將背景任務封裝成服務類別：例如 class EmailService:，讓 DI 可以直接注入服務物件。
3. 使用 BaseSettings 統一管理環境變數，在依賴中返回設定物件，避免硬編碼。
4. 設定適當的超時與錯誤處理：在背景任務內捕捉例外，避免未處理的錯誤導致進程崩潰。
5. 監控與日誌：將背景任務的成功/失敗寫入統一日誌或監控系統（如 Prometheus），有助於問題排查。

---

實際應用場景

場景	為什麼需要 background tasks + dependencies
使用者註冊後寄送驗證信	註冊請求需要立即回應，發信可延後；依賴可提供 SMTP 客戶端或第三方 Email Service。
上傳檔案後產生縮圖	圖片處理耗時，背景任務負責縮圖、上傳至雲端儲存；依賴提供雲端儲存 client。
訂單成立後發送 Slack/Line 通知	商業流程需要即時提醒，背景任務呼叫外部 webhook；依賴注入 webhook URL 與 HTTP client。
日誌或審計紀錄寫入資料庫	交易量大時同步寫入會拖慢回應，背景任務批次寫入；依賴提供同一個 DB session，減少連線開銷。
批次報表產生 & 電子郵件寄送	使用者請求產生報表，報表產生可能需要數分鐘，背景任務完成後自動寄送；依賴提供報表產生服務與 Email 服務。

---

總結

• BackgroundTasks 讓我們在回應送出後，非同步執行耗時工作，提升 API 的即時性。
• Dependency Injection 為背景任務提供 資源、設定與驗證，避免在每個任務內重複建立連線或硬編碼。
• 結合兩者，我們可以寫出 乾淨、可測試、可擴充 的程式碼，且在實務專案中輕鬆處理郵件、訊息、審計、檔案處理等常見需求。

掌握了「background tasks + dependencies」的技巧後，你將能在 FastAPI 中構建更具彈性與可維護性的服務，從而在高併發與業務複雜度日益提升的環境中，保持開發效率與系統穩定性。祝你寫程式寫得開心，API 服務跑得順暢 🚀