本文 AI 產出,尚未審核

FastAPI

單元:依賴注入系統(Dependency Injection)

主題:函式依賴注入


簡介

在構建 Web API 時,維護性可測試性可重用性 常常是最難以同時兼顧的三大挑戰。FastAPI 之所以能在短時間內成為 Python 生態系的明星框架,很大程度上得益於它內建的 依賴注入(Dependency Injection, DI) 機制。透過 DI,我們可以把「取得資料」或「建立資源」等副作用的程式碼,從路由函式本身抽離出來,變成可獨立測試、可在多個端點共享的 函式依賴

本篇文章聚焦於 函式依賴注入 的核心概念與實作方式,從最基本的「把函式當作參數」開始,一路帶到 依賴層級自動清理異步依賴 等進階技巧。即使你是剛接觸 FastAPI 的新手,也能在閱讀完後立刻把這套機制運用到自己的專案中;若你已經有一定開發經驗,則能從最佳實踐與常見陷阱中得到提升。


核心概念

1. 什麼是函式依賴?

在 FastAPI 中,函式依賴 就是「一個普通的 Python 函式」,它的返回值會被注入(injected)到另一個需要它的函式(通常是路由處理函式)中。FastAPI 會在每一次請求時自動執行依賴函式,取得返回值後作為參數傳遞給目標函式。

關鍵點:依賴函式本身可以是同步的 def,也可以是異步的 async def;只要符合 呼叫簽名(參數與返回型別)即可。

2. 為什麼要使用函式依賴?

好處 說明
解耦 讓路由函式只關心「業務邏輯」,不必自行處理 DB 連線、驗證或設定讀取等雜務。
可測試 依賴函式可以在單元測試時被 override(覆寫),不需要真的連到資料庫或外部服務。
重用 同一個依賴可以在多個路由、甚至在子應用(sub‑app)之間共享。
自動清理 使用 yield 的依賴會在請求結束後自動執行清理程式(例如關閉 DB 連線)。
型別提示 配合 Pydantic 與 Python 型別註解,FastAPI 能提供自動文件與驗證。

3. 基本語法

from fastapi import FastAPI, Depends

app = FastAPI()

def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
    """
    這是一個簡單的依賴函式,負責統一處理分頁與搜尋參數。
    """
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
async def read_items(params: dict = Depends(common_parameters)):
    """
    FastAPI 會先呼叫 common_parameters,取得回傳的 dict,
    再把它注入到 `params` 參數中。
    """
    return {"message": "取得 items", "params": params}
  • Depends(common_parameters) 告訴 FastAPI「這個參數的值必須由 common_parameters 產生」。
  • 無需手動呼叫 common_parameters,FastAPI 會在每一次請求時自動執行。

4. 依賴的層級(Nested Dependencies)

依賴本身也可以依賴其他函式,形成 層級結構。這讓我們可以把「最底層」的資源(如資料庫連線)封裝起來,再在上層組合成更高層的服務物件。

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session

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

def get_db() -> Session:
    """
    產生一個 SQLAlchemy Session,使用 `yield` 讓 FastAPI 在請求結束後自動關閉。
    """
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def get_current_user(db: Session = Depends(get_db)):
    """
    以 DB Session 為基礎,取得目前的使用者(示意)。
    """
    # 假設有一個 User model 與 token 驗證
    user = db.query(User).filter(User.id == 1).first()
    return user

@app.get("/profile/")
def read_profile(user: User = Depends(get_current_user)):
    return {"username": user.username, "email": user.email}
  • get_current_user 依賴 get_db,形成 兩層依賴
  • /profile/ 被呼叫時,FastAPI 會依序執行 get_dbget_current_userread_profile

5. 異步依賴(Async Dependency)

如果依賴本身需要執行 I/O(例如呼叫外部 API、Redis、MongoDB),使用 async def 能讓整個請求流程保持非阻塞。

import httpx

async def get_weather(city: str):
    """
    透過非同步 HTTP 請求取得天氣資訊。
    """
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"https://api.weather.com/v3/weather/{city}")
        resp.raise_for_status()
        return resp.json()

@app.get("/weather/{city}")
async def weather(city: str, data: dict = Depends(get_weather)):
    return {"city": city, "weather": data}
  • get_weather異步依賴,FastAPI 會在事件迴圈中執行 await,不會阻塞其他請求。

6. 使用 Depends 於類別(Class‑based Dependency)

有時候我們想把依賴封裝成類別,尤其是當需要保存狀態或多個方法時。FastAPI 允許把類別本身作為依賴,或把類別的實例方法作為依賴。

class AuthService:
    def __init__(self, token: str = Depends(oauth2_scheme)):
        self.token = token

    def verify(self):
        # 假設驗證 token 並返回使用者 ID
        if self.token == "secret-token":
            return 42
        raise HTTPException(status_code=401, detail="Invalid token")

def get_auth_service(auth: AuthService = Depends()):
    """
    直接返回 AuthService 實例,FastAPI 會自動解析其建構子中的依賴。
    """
    return auth

@app.get("/secure-data/")
def secure_endpoint(user_id: int = Depends(lambda s=Depends(get_auth_service): s.verify())):
    return {"user_id": user_id, "data": "這是保護的資源"}
  • AuthService 的建構子使用 Depends(oauth2_scheme) 取得 token。
  • 透過 Depends(get_auth_service) 取得已初始化的服務實例,再呼叫其方法。

7. 覆寫依賴(Dependency Override)

在測試或臨時環境中,我們常需要 替換 真實的依賴。FastAPI 提供 app.dependency_overrides 讓開發者在程式碼層面或測試套件中輕鬆覆寫。

# 真實環境的依賴
def get_current_user(db: Session = Depends(get_db)):
    # 讀取資料庫取得使用者
    ...

# 測試環境的假資料
def fake_user():
    class User:
        id = 1
        username = "test_user"
    return User()

# 在測試前設定覆寫
app.dependency_overrides[get_current_user] = fake_user

# 測試結束後記得清除
app.dependency_overrides.clear()
  • 覆寫後,所有使用 Depends(get_current_user) 的路由都會得到 fake_user() 的回傳值,讓測試不必依賴資料庫。

常見陷阱與最佳實踐

陷阱 說明 解決方案
依賴函式內部寫死全域變數 會破壞測試的可預測性,因為每次請求都會共用同一個實例。 使用 Depends 注入需要的資源,或在函式內部 建立(而非引用)本地變數。
忘記使用 yield 釋放資源 若依賴產生了資料庫連線、檔案句柄等,未使用 yield 會導致資源泄漏。 將需要清理的程式寫在 finally 區塊,或改用 async with/with 結構。
過度嵌套依賴 依賴層級過深會讓除錯變得困難,且每層都會產生一次呼叫成本。 保持依賴的 單一職責,必要時可把多個相關依賴合併成一個服務類別。
在依賴函式內直接拋出 HTTPException 會讓錯誤訊息在不同層級混雜,失去統一的錯誤處理策略。 建議在依賴內只回傳結果,在路由層或全局例外處理器中統一拋出 HTTPException
使用非序列化的物件作為依賴回傳值 FastAPI 的自動文件生成只能對 Pydantic 模型或基本型別做說明。 若要在 OpenAPI 文件中顯示,請使用 Pydantic BaseModel 包裝回傳資料。

最佳實踐清單

  1. 保持函式依賴簡潔:每個依賴只負責一件事(例如「取得 DB Session」或「驗證 Token」)。
  2. 使用型別註解:讓 IDE 與 FastAPI 都能正確推斷參數與返回值。
  3. 盡量使用 yield:即使目前不需要清理,也養成寫法,未來擴充時不會遺漏。
  4. 在測試前覆寫依賴:使用 dependency_overrides,確保測試環境與正式環境徹底分離。
  5. 避免在依賴內部直接存取 request body:若需要請求資料,使用 Request 物件作為依賴參數,而不是在全域變數中讀取。

實際應用場景

1. 多租戶(Multi‑Tenant)系統

在 SaaS 平台中,每個租戶都有自己的資料庫或資料表前綴。利用依賴注入,我們可以把「根據請求的 Header 判斷租戶」的邏輯抽成一個依賴,然後在每個需要 DB Session 的路由中自動帶入正確的連線資訊。

def get_tenant_id(request: Request):
    tenant_id = request.headers.get("X-Tenant-ID")
    if not tenant_id:
        raise HTTPException(status_code=400, detail="Missing X-Tenant-ID")
    return tenant_id

def get_tenant_db(tenant_id: str = Depends(get_tenant_id)):
    # 假設每個租戶都有獨立的 SQLite 檔案
    db_url = f"sqlite:///./{tenant_id}.db"
    engine = create_engine(db_url, connect_args={"check_same_thread": False})
    SessionLocal = sessionmaker(bind=engine)
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

好處:所有路由只需要 db: Session = Depends(get_tenant_db),租戶切換全自動,且測試時只要覆寫 get_tenant_id 即可模擬不同租戶。

2. 跨服務的認證與授權

在微服務架構裡,API Gateway 會先驗證 JWT,然後把使用者資訊寫入 Request State。接下來的服務只需要一個 get_current_user 依賴即可取得已驗證的使用者,無需再次解析 token。

def get_current_user(state: dict = Depends(lambda request: request.state)):
    user = state.get("user")
    if not user:
        raise HTTPException(status_code=401, detail="Unauthenticated")
    return user
  • 與 Gateway 整合:Gateway 把 request.state.user = decoded_user,子服務只負責 Depends(get_current_user)

3. 實作「限流」與「緩存」

使用 Redis 作為分布式限流與快取後端,將相關邏輯封裝成依賴,讓每個路由只需要 rate_limiter: RateLimiter = Depends() 即可。

import aioredis

class RateLimiter:
    def __init__(self, redis: aioredis.Redis, limit: int = 10):
        self.redis = redis
        self.limit = limit

    async def __call__(self, request: Request):
        ip = request.client.host
        key = f"rl:{ip}"
        current = await self.redis.incr(key)
        if current == 1:
            await self.redis.expire(key, 60)  # 1 分鐘內的計數
        if current > self.limit:
            raise HTTPException(status_code=429, detail="Too Many Requests")

async def get_redis() -> aioredis.Redis:
    return await aioredis.from_url("redis://localhost")

def get_rate_limiter(redis: aioredis.Redis = Depends(get_redis)):
    return RateLimiter(redis)

@app.get("/resource/")
async def protected_endpoint(limiter: RateLimiter = Depends(get_rate_limiter)):
    return {"msg": "你沒有被限流!"}
  • 優點:限流邏輯與路由分離,且可以在測試環境把 get_redis 覆寫成 mock

總結

函式依賴注入是 FastAPI 最具威力的特性之一,它讓業務程式碼基礎設施(資料庫、認證、緩存、外部 API)徹底分離。透過 Depends、層級依賴、yield 清理與異步支援,我們能夠:

  • 提升可測試性:只要覆寫依賴,就能在單元測試中使用假資料。
  • 加速開發:共用的資源(DB Session、Auth Service 等)只寫一次,所有路由自動受惠。
  • 確保資源安全:使用 yieldasync with,FastAPI 會自動在請求結束時釋放資源。
  • 保持程式碼可讀:每個依賴只負責單一職責,讓程式結構清晰、易於維護。

在實務專案中,建議先從 最簡單的函式依賴(如共用參數、簡易驗證)開始,逐步擴展到 層級依賴類別服務異步資源,最後根據需求加入 依賴覆寫 以支援測試與環境切換。只要遵循本文的 最佳實踐,你就能在 FastAPI 中建立高品質、易維護且具備彈性伸縮的 API 服務。

祝開發愉快 🎉