本文 AI 產出,尚未審核

FastAPI 與 IoC 思維:深入探討依賴注入系統(Dependency Injection)

簡介

在現代的 Web 框架中,依賴注入(Dependency Injection,簡稱 DI) 已成為提升程式可測試性、可維護性與可重用性的關鍵技術。FastAPI 以其「宣告式」的路由與自動產生 OpenAPI 文件聞名,同時也內建了功能完整且易於使用的 DI 系統,讓開發者可以以 控制反轉(Inversion of Control,IoC) 的思維,將外部資源(資料庫、快取、第三方服務等)「注入」到路由函式或是背景任務中。

本篇文章將從 IoC 思維的概念 出發,說明 FastAPI 如何實作依賴注入、提供多種實用範例,並分享常見陷阱與最佳實踐,協助讀者在實務專案中善用這套機制,寫出更乾淨、更易測試的程式碼。


核心概念

1. 為什麼要使用 IoC / DI?

傳統寫法 IoC/DI 寫法
def get_user(id: int): db = create_engine(); conn = db.connect(); … def get_user(id: int, db: Session = Depends(get_db)):
依賴硬耦合於函式內部,測試時需自行 mock 或建立實體 DB。 依賴抽象化為「可注入」的參數,測試時只要提供不同的實例即可。
  • 降低耦合度:業務邏輯不再直接建立或管理資源。
  • 提升測試便利性:可以在測試環境注入 FakeStubIn‑Memory 版本。
  • 統一資源管理:如資料庫連線池、Redis 客戶端只需要在一個地方建立,其他地方只要 Depends 即可取得。

2. FastAPI 的 Depends

FastAPI 透過 fastapi.Depends 來宣告「此參數是一個依賴」,框架會在請求進來時自動解析、執行相對應的「依賴提供者」函式,並把回傳值注入到目標函式的參數中。

from fastapi import Depends, FastAPI

app = FastAPI()

def common_parameters(q: str = None, limit: int = 10):
    """提供共用查詢參數的依賴函式"""
    return {"q": q, "limit": limit}

@app.get("/items/")
def read_items(params: dict = Depends(common_parameters)):
    return {"message": "使用共用參數", "params": params}
  • common_parameters 只負責組裝參數,不涉及任何業務邏輯
  • Depends 讓 FastAPI 在每次請求時自動呼叫 common_parameters,並把結果傳入 read_items

3. 依賴的生命週期(Scope)

FastAPI 支援三種主要的生命週期:

Scope 說明 常見使用情境
request(預設) 每個 HTTP 請求產生一次實例。 DB Session、認證資訊
session 針對 WebSocket 連線或長連線維持同一實例。 WebSocket 共享資源
singleton 應用程式啟動時建立一次,之後共用同一實例。 設定檔、外部 API 客戶端

使用 Depends 時可以透過 @lru_cacheDepends(..., use_cache=False) 或自行實作 ContextVar 來控制快取行為。

from functools import lru_cache

@lru_cache()
def get_settings():
    """Singleton 依賴:整個應用只建立一次設定物件"""
    return Settings()          # 假設 Settings 為 pydantic.BaseSettings

@app.get("/config")
def read_config(settings: Settings = Depends(get_settings)):
    return settings.dict()

4. 多層依賴(Nested Dependencies)

依賴本身也可以依賴其他依賴,形成 依賴樹。FastAPI 會自動解析整棵樹,確保每個節點只被呼叫一次(除非 use_cache=False)。

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

def get_current_user(db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)):
    user = verify_token(token, db)
    return user

@app.get("/profile")
def read_profile(user: User = Depends(get_current_user)):
    return {"username": user.username}

此例中,read_profile 只需要 User 物件,FastAPI 會先建立 DB Session、驗證 token,最後把 User 注入。


程式碼範例(實用示範)

以下提供 5 個 常見且實務導向的範例,說明如何在 FastAPI 中運用 DI。

範例 1:資料庫 Session(Request Scope)

# db.py
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:
    """每個請求建立一個 DB Session,請求結束自動關閉"""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
# main.py
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from . import models, db

app = FastAPI()

@app.post("/users/")
def create_user(user: models.UserCreate, db: Session = Depends(db.get_db)):
    db_user = models.User(**user.dict())
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

重點:使用 yield 讓 FastAPI 知道這是一個「生成器依賴」,框架會在請求結束時自動執行 finally 區塊,確保資源釋放。


範例 2:認證與授權(多層依賴)

# auth.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from sqlalchemy.orm import Session
from .db import get_db
from . import crud, schemas

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="無效的認證資訊",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, "SECRET_KEY", algorithms=["HS256"])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = crud.get_user_by_username(db, username=username)
    if user is None:
        raise credentials_exception
    return user
# main.py
@app.get("/me")
def read_me(current_user: schemas.User = Depends(auth.get_current_user)):
    return {"username": current_user.username, "email": current_user.email}

技巧oauth2_scheme 本身就是一個依賴,將它與 get_db 結合,即可在同一函式內同時取得 token 與 DB 連線。


範例 3:設定檔(Singleton Scope)

# config.py
from pydantic import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    app_name: str = "FastAPI Demo"
    admin_email: str
    items_per_user: int = 50

    class Config:
        env_file = ".env"

@lru_cache()
def get_settings() -> Settings:
    return Settings()
# main.py
from fastapi import Depends, FastAPI
from .config import get_settings, Settings

app = FastAPI()

@app.get("/info")
def get_info(settings: Settings = Depends(get_settings)):
    return {"app_name": settings.app_name, "admin": settings.admin_email}

說明@lru_cacheget_settings 只在第一次呼叫時讀取 .env,之後直接回傳快取結果,達到 singleton 效果。


範例 4:快取服務(Redis)與自訂依賴

# cache.py
import aioredis
from functools import lru_cache

REDIS_URL = "redis://localhost"

@lru_cache()
def get_redis() -> aioredis.Redis:
    return aioredis.from_url(REDIS_URL, decode_responses=True)

async def get_cached_value(key: str, redis: aioredis.Redis = Depends(get_redis)):
    value = await redis.get(key)
    return value
# main.py
from fastapi import Depends, FastAPI

app = FastAPI()

@app.get("/cache/{key}")
async def read_cache(key: str, value: str = Depends(cache.get_cached_value)):
    return {"key": key, "value": value or "未命中"}

注意:Redis 客戶端是 非同步,所以依賴函式與路由都必須使用 async def


範例 5:測試時注入 Fake 服務

# test_conftest.py
import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.dependencies import get_db

@pytest.fixture
def client():
    # 使用 SQLite 記憶體資料庫作為測試用 DB
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    engine = create_engine("sqlite:///:memory:")
    TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

    # 建立測試用的依賴
    def override_get_db():
        db = TestingSessionLocal()
        try:
            yield db
        finally:
            db.close()

    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()
def test_create_user(client):
    response = client.post("/users/", json={"username":"alice","email":"a@example.com"})
    assert response.status_code == 200
    assert response.json()["username"] == "alice"

關鍵app.dependency_overrides 讓測試環境可以 替換任何依賴,不必改動原始程式碼,達到乾淨的單元測試。


常見陷阱與最佳實踐

陷阱 可能的後果 解決方式 / 最佳實踐
依賴函式忘記 yield,導致資源不會自動釋放 連線池耗盡、記憶體洩漏 使用 try…finally 包住 yield,或直接回傳非生成器物件(若不需要釋放)
多層依賴快取失效use_cache=False 同一請求內重複建立 DB Session,效能下降 預設使用快取,只有在需要「每次都重新建立」的情況才關閉快取
將全域單例寫成 request scope 每次請求都重新建立昂貴物件(如大型模型) 針對不變的資源使用 @lru_cache 或自行管理 singleton
依賴函式內使用同步阻塞 I/O(如 requests 事件迴圈被阻塞,導致整體吞吐量下降 改用 httpx.AsyncClient 或其他非同步庫
測試時忘記清除 dependency_overrides 測試之間互相干擾,產生莫名錯誤 在測試結束後 app.dependency_overrides.clear(),或使用 fixture 統一管理

最佳實踐

  1. 明確分層:將「外部資源建立」與「業務邏輯」徹底分離,依賴只負責提供資源。
  2. 使用型別提示:FastAPI 會根據型別自動產生 OpenAPI,讓 IDE 以及文件更友好。
  3. 最小化依賴樹深度:過深的依賴樹會讓除錯變困難,適度抽象即可。
  4. startup / shutdown 事件中建立長期資源(如模型載入),避免每次請求重新載入。
  5. 測試時善用 dependency_overrides,保持測試與正式環境的程式碼一致性。

實際應用場景

場景 為何需要 DI 典型實作
多租戶 SaaS 系統 每個租戶需要不同的資料庫連線或快取命名空間 Depends(get_tenant_db) 中根據請求 Header 解析租戶 ID,返回對應的 Session
機器學習模型服務 模型載入成本高,需在應用啟動時載入一次,並在多個路由共享 使用 @lru_cache 建立 load_model(),在路由中 model = Depends(load_model)
第三方 API 客戶端 認證 token 需要定期刷新,且多個端點會共用同一 client 建立 get_external_client(),內部處理 token 自動更新,依賴注入給所有需要呼叫外部服務的路由
背景任務與 WebSocket 同一個連線需要共享資源(如 Redis Pub/Sub) 使用 session scope 的依賴,確保 WebSocket 連線期間資源不被釋放
測試環境切換 測試需要使用 SQLite 記憶體或 Mock 服務 透過 app.dependency_overrides 替換 get_dbget_redis 等依賴

總結

FastAPI 的依賴注入系統以 宣告式自動解析 為核心,讓開發者可以把 「取得資源」 的工作交給框架,自己只專注於 業務邏輯。透過 IoC 思維,我們將資源的建立與使用解耦,從而:

  • 提升程式可測試性:測試時只要替換依賴即可,不必改動原始程式碼。
  • 統一資源管理:生命週期(request、session、singleton)清晰可控,避免資源洩漏。
  • 加速開發:共用參數、認證、設定等常見需求只要寫一次依賴函式,即可在多個路由間重用。

掌握了 Dependsyield@lru_cache 以及 dependency_overrides,你就能在 FastAPI 中自如地運用 IoC,構建出乾淨、可維護、易測試的現代 Web API。祝你開發順利,快把這套思維應用到下一個專案吧!