本文 AI 產出,尚未審核

FastAPI 教學:依賴注入系統(Dependency Injection)— 多層依賴(Nested Dependencies)


簡介

FastAPI 中,依賴注入(Dependency Injection,簡稱 DI)是讓程式碼保持乾淨、可測試、可重用的關鍵機制。單一依賴已經能夠把資料庫連線、認證、設定檔等共用資源抽離出來,但在實際專案裡,往往會出現 「多層依賴」(nested dependencies)的情況:一個依賴本身又依賴其他依賴。

掌握多層依賴的寫法與注意事項,能讓你:

  • 降低耦合度:每個功能只關心自己需要的資源,其他細節交給更上層的依賴負責。
  • 提升測試效率:只要替換最外層的依賴,即可在單元測試中模擬整個依賴樹。
  • 簡化程式流程:透過 FastAPI 自動解析依賴樹,開發者不必手動傳遞參數。

本篇文章將從概念說明、實作範例、常見陷阱與最佳實踐,到實務應用場景,一步步帶你熟悉 多層依賴 的寫法與技巧。


核心概念

1. 什麼是「多層依賴」?

在 FastAPI 中,依賴 是一個可呼叫的物件(函式、類別或 async 函式),FastAPI 會在路由被呼叫時自動執行它,並把返回值注入到路由函式的參數中。
A 依賴 B,而 B 又依賴 C 時,就形成了「多層依賴」:

Route → A → B → C

FastAPI 會自動解析整條鏈,從最底層的 C 開始執行,逐層往上返回結果,最後把 A 的返回值注入到路由。

2. 為什麼要使用多層依賴?

場景 單層依賴的缺點 多層依賴的好處
資料庫 + 事務 每個路由都必須自行取得 Session,且事務管理散落在多處 Session 放在最底層,事務邏輯封裝於上層依賴,路由只需要注入「服務」
認證 + 權限檢查 每個需要權限的路由都要重複寫 current_user + permission_check 認證權限 分別抽成兩層依賴,組合成「授權服務」供路由使用
設定 + 外部 API 客戶端 每個呼叫外部 API 的路由都要自行讀取設定檔與建立客戶端 設定 放在最底層,API 客戶端 依賴設定,最外層依賴只提供「已配置好的客戶端」

3. FastAPI 解析依賴的流程

  1. 解析路由函式的參數,找出標註為 Depends(...) 的依賴。
  2. 遞迴解析:對每個依賴再次檢查其參數是否為 Depends,直到最底層。
  3. 執行順序:從最底層依賴往上執行,並將返回值緩存於請求的生命週期(依賴的 scope 決定)。
  4. 注入結果:把每層返回的物件注入到上層的參數,最終注入到路由函式。

小技巧:如果某層依賴的返回值需要在同一次請求中共用,請使用 Depends(..., use_cache=True)(預設即為 True),FastAPI 會自動快取結果,避免重複呼叫。


程式碼範例

以下示範 5 個常見的多層依賴情境,全部採用 Python(FastAPI 原生語言),並加上詳細註解說明。

範例 1️⃣:設定檔 → DB Session → Repository → Service → Route

# app/config.py
from pydantic import BaseSettings

class Settings(BaseSettings):
    database_url: str = "sqlite:///./test.db"

    class Config:
        env_file = ".env"

def get_settings() -> Settings:
    """最底層依賴:提供全域設定物件"""
    return Settings()
# app/db.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from .config import get_settings, Settings

def get_engine(settings: Settings = Depends(get_settings)):
    """依賴 Settings,建立資料庫引擎"""
    return create_engine(settings.database_url, connect_args={"check_same_thread": False})

def get_db(engine=Depends(get_engine)):
    """提供 SQLAlchemy Session,使用 request scope 快取"""
    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    db: Session = SessionLocal()
    try:
        yield db
    finally:
        db.close()
# app/repositories/user.py
from sqlalchemy.orm import Session
from ..models import User

class UserRepository:
    def __init__(self, db: Session):
        self.db = db

    def get_by_id(self, user_id: int) -> User | None:
        return self.db.query(User).filter(User.id == user_id).first()
# app/services/user.py
from fastapi import Depends
from ..repositories.user import UserRepository
from ..db import get_db
from sqlalchemy.orm import Session

def get_user_repository(db: Session = Depends(get_db)):
    """依賴 DB Session,回傳 Repository 實例"""
    return UserRepository(db)

class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo

    def get_user(self, user_id: int):
        user = self.repo.get_by_id(user_id)
        if not user:
            raise ValueError("User not found")
        return user

def get_user_service(repo: UserRepository = Depends(get_user_repository)):
    """最外層依賴:提供已注入 Repository 的 Service"""
    return UserService(repo)
# app/main.py
from fastapi import FastAPI, Depends, HTTPException
from .services.user import get_user_service, UserService

app = FastAPI()

@app.get("/users/{user_id}")
def read_user(user_id: int, service: UserService = Depends(get_user_service)):
    """
    路由只需要注入最外層的 Service,
    內部所有依賴(設定、DB、Repository)皆由 FastAPI 自動解析。
    """
    try:
        return service.get_user(user_id)
    except ValueError as e:
        raise HTTPException(status_code=404, detail=str(e))

重點:從 SettingsEngineSessionRepositoryService,每層只關心自己的職責,路由保持最簡潔。


範例 2️⃣:認證 → 權限檢查 → 路由

# app/auth.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def get_current_user(token: str = Depends(oauth2_scheme)):
    """
    假設 token 內含使用者 ID,實際情況會驗證 JWT 或其他方式。
    """
    if token != "valid-token":
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
    # 這裡直接回傳模擬的使用者資料
    return {"username": "alice", "roles": ["admin", "user"]}
# app/permissions.py
from fastapi import Depends, HTTPException, status

def require_role(role: str):
    """
    高階函式,返回一個依賴函式,用於檢查使用者是否具備指定角色。
    """
    def role_checker(current_user: dict = Depends(get_current_user)):
        if role not in current_user["roles"]:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Missing role: {role}"
            )
        return current_user
    return role_checker
# app/main.py (續)
@app.get("/admin/dashboard")
def admin_dashboard(user: dict = Depends(require_role("admin"))):
    """
    只要使用 `require_role("admin")`,就自動完成「認證」+「權限」的多層依賴。
    """
    return {"msg": f"Welcome {user['username']}! This is the admin dashboard."}

技巧require_role工廠函式(factory),可動態產生不同角色的依賴,讓權限檢查保持彈性。


範例 3️⃣:外部 API 客戶端(依賴設定)→ 服務層→ 路由

# app/external/config.py
from pydantic import BaseSettings

class ExternalAPISettings(BaseSettings):
    base_url: str = "https://api.example.com"
    api_key: str = "default-key"

    class Config:
        env_prefix = "EXTERNAL_"

def get_external_settings() -> ExternalAPISettings:
    return ExternalAPISettings()
# app/external/client.py
import httpx
from .config import get_external_settings, ExternalAPISettings
from fastapi import Depends

def get_http_client(settings: ExternalAPISettings = Depends(get_external_settings)):
    """
    建立一個共享的 httpx.AsyncClient,設定好 base_url 與認證頭。
    """
    client = httpx.AsyncClient(
        base_url=settings.base_url,
        headers={"Authorization": f"Bearer {settings.api_key}"}
    )
    return client
# app/external/service.py
from fastapi import Depends
from .client import get_http_client
import httpx

class WeatherService:
    def __init__(self, client: httpx.AsyncClient):
        self.client = client

    async def get_current_weather(self, city: str) -> dict:
        resp = await self.client.get(f"/weather?city={city}")
        resp.raise_for_status()
        return resp.json()

def get_weather_service(client: httpx.AsyncClient = Depends(get_http_client)):
    return WeatherService(client)
# app/main.py (續)
@app.get("/weather/{city}")
async def weather(city: str, svc: WeatherService = Depends(get_weather_service)):
    """
    路由只需要注入最外層的 WeatherService。
    內部的設定與 http client 都已在多層依賴中完成。
    """
    return await svc.get_current_weather(city)

說明:設定檔只讀一次(快取),httpx.AsyncClient 也只建立一次,提升效能。


範例 4️⃣:事務 (Transaction) 依賴 → 服務 → 路由

# app/db_transaction.py
from sqlalchemy.orm import Session
from .db import get_db
from fastapi import Depends

def get_transaction(db: Session = Depends(get_db)):
    """
    包裝一個簡易的事務管理器,使用者只需要在服務層 `with` 使用。
    """
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
        raise
# app/services/order.py
from fastapi import Depends
from ..db_transaction import get_transaction
from sqlalchemy.orm import Session

class OrderService:
    def __init__(self, db: Session):
        self.db = db

    def create_order(self, user_id: int, product_id: int):
        # 假設有 Order、Inventory 兩個 model,以下僅示意
        order = {"user_id": user_id, "product_id": product_id}
        # self.db.add(order)  # 實際寫入 DB
        # 同時扣庫存...
        return order

def get_order_service(db: Session = Depends(get_transaction)):
    return OrderService(db)
# app/main.py (續)
@app.post("/orders")
def create_order(user_id: int, product_id: int,
                 svc: OrderService = Depends(get_order_service)):
    """
    只要呼叫 `svc.create_order`,事務的 commit / rollback 已由
    `get_transaction` 自動處理,路由不必關心。
    """
    return svc.create_order(user_id, product_id)

重點:把 事務管理 抽成最底層依賴,讓服務層只寫業務邏輯,錯誤時自動 rollback。


範例 5️⃣:測試時的依賴替換(Mock 多層依賴)

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.services.user import UserService

class MockUserService:
    def get_user(self, user_id: int):
        return {"id": user_id, "username": "mock_user"}

# 替換最外層的依賴
app.dependency_overrides[UserService] = lambda: MockUserService()

@pytest.fixture
def client():
    return TestClient(app)
# tests/test_user.py
def test_read_user(client):
    response = client.get("/users/1")
    assert response.status_code == 200
    data = response.json()
    assert data["username"] == "mock_user"

說明:只要在測試環境把最外層依賴(UserService)換成 mock,整條依賴樹自動使用 mock,無需逐層替換。


常見陷阱與最佳實踐

陷阱 可能的問題 解決方案或最佳實踐
依賴快取失效 Depends(..., use_cache=False) 中每次請求都重新建立資源(如 DB 連線),導致效能下降。 預設使用 use_cache=True,除非確實需要每次重新建立(例如測試環境的隨機值)。
循環依賴 A 依賴 B,B 又依賴 A,會在啟動時拋出 RecursionError 重新設計依賴圖,將共同需求抽成第三層(C),或使用 依賴工廠(factory)延遲呼叫。
過度嵌套 超過 4–5 層依賴會讓除錯變得困難。 盡量保持每層職責單一,必要時把「服務」與「業務流程」合併到同一層,或使用 Command Pattern 包裝多個服務。
不一致的 Scope 有的依賴使用 request scope,有的使用 singleton,但在同一條鏈中混用會造成資源共享不當。 明確決定每個依賴的生命週期:設定、客戶端 → singleton;DB Session、事務 → request
測試時忘記恢復 overrides app.dependency_overrides 若未在測試結束後清除,會影響後續測試。 使用 pytest.fixturefinalizertry/finally 在測試後清空 dependency_overrides

最佳實踐小結

  1. 分層明確:設定 → 客戶端 → Repository → Service → Route。每層只做一件事。
  2. 使用 Depends 產生工廠:像 require_role 那樣的高階函式,讓依賴更具彈性。
  3. 快取資源:大量建立的物件(如 httpx.AsyncClient、資料庫引擎)應在 singleton 層快取。
  4. 合理設定 Scopesingleton(應用啟動時一次) vs request(每次請求一次)。
  5. 測試友好:只要在測試中覆寫最外層依賴,即可輕鬆 mock 整條依賴樹。

實際應用場景

場景 多層依賴的實作方式
微服務間的認證與授權 1. Settings 讀取 JWT 公鑰
2. JWTVerifier 依賴 Settings
3. CurrentUser 依賴 JWTVerifier
4. PermissionChecker 依賴 CurrentUser
多資料庫(讀寫分離) 1. Settings 包含 master_urlreplica_url
2. EngineMasterEngineReplica 各自依賴 Settings
3. SessionMasterSessionReplica 分別依賴對應 Engine
4. Repository 依賴兩個 Session,根據需求選擇讀/寫
長連線的 WebSocket 1. SettingsRedisClient(pub/sub)
2. MessageBroker 依賴 RedisClient
3. WebSocketManager 依賴 MessageBroker,在每條連線建立時注入
資料批次處理(背景工作) 1. SettingsCeleryApp(或 BackgroundTasks
2. TaskRepository 依賴 DB Session
3. EmailSender 依賴 SMTP 設定
4. UserReportTask 依賴 TaskRepository + EmailSender,最外層由 Celery worker 呼叫
多租戶 SaaS 平台 1. TenantResolver 依賴請求 Header / JWT
2. TenantSettings 依賴 TenantResolver(根據租戶讀取不同 DB URL)
3. EngineSession 依賴 TenantSettings,形成「租戶感知」的資料層

這些場景的共同點是:每個層級只負責自己的設定或資源,最外層的服務或路由只需要注入最終的「業務物件」,大幅提升程式碼的可讀性與可維護性。


總結

  • 多層依賴 是 FastAPI 強大 DI 系統的核心特性之一,讓開發者能把設定、連線、事務、認證、授權等功能以 分層可重用 的方式組合。
  • 正確使用 Dependsuse_cache 以及 生命週期(scope),可以避免資源浪費與循環依賴的陷阱。
  • 透過 工廠函式(如 require_role)與 測試覆寫dependency_overrides),我們可以在保持程式碼乾淨的同時,快速適應不同的業務需求與測試情境。
  • 真正的實務價值在於:路由只需要注入最外層服務,底層的所有依賴都交給 FastAPI 自動解決,讓專案在擴充與維護時更具彈性。

掌握了多層依賴的寫法,你就能在 FastAPI 專案中打造 乾淨、可測、可擴充 的架構,從小型原型一路走向企業級服務。祝開發順利,期待在你的專案裡看到這些技巧的身影! 🚀