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 解析依賴的流程
- 解析路由函式的參數,找出標註為
Depends(...)的依賴。 - 遞迴解析:對每個依賴再次檢查其參數是否為
Depends,直到最底層。 - 執行順序:從最底層依賴往上執行,並將返回值緩存於請求的生命週期(依賴的
scope決定)。 - 注入結果:把每層返回的物件注入到上層的參數,最終注入到路由函式。
小技巧:如果某層依賴的返回值需要在同一次請求中共用,請使用
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))
重點:從
Settings→Engine→Session→Repository→Service,每層只關心自己的職責,路由保持最簡潔。
範例 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.fixture 的 finalizer 或 try/finally 在測試後清空 dependency_overrides。 |
最佳實踐小結
- 分層明確:設定 → 客戶端 → Repository → Service → Route。每層只做一件事。
- 使用
Depends產生工廠:像require_role那樣的高階函式,讓依賴更具彈性。 - 快取資源:大量建立的物件(如
httpx.AsyncClient、資料庫引擎)應在 singleton 層快取。 - 合理設定 Scope:
singleton(應用啟動時一次) vsrequest(每次請求一次)。 - 測試友好:只要在測試中覆寫最外層依賴,即可輕鬆 mock 整條依賴樹。
實際應用場景
| 場景 | 多層依賴的實作方式 |
|---|---|
| 微服務間的認證與授權 | 1. Settings 讀取 JWT 公鑰2. JWTVerifier 依賴 Settings3. CurrentUser 依賴 JWTVerifier4. PermissionChecker 依賴 CurrentUser |
| 多資料庫(讀寫分離) | 1. Settings 包含 master_url、replica_url2. EngineMaster、EngineReplica 各自依賴 Settings3. SessionMaster、SessionReplica 分別依賴對應 Engine4. Repository 依賴兩個 Session,根據需求選擇讀/寫 |
| 長連線的 WebSocket | 1. Settings → RedisClient(pub/sub)2. MessageBroker 依賴 RedisClient3. WebSocketManager 依賴 MessageBroker,在每條連線建立時注入 |
| 資料批次處理(背景工作) | 1. Settings → CeleryApp(或 BackgroundTasks)2. TaskRepository 依賴 DB Session3. EmailSender 依賴 SMTP 設定4. UserReportTask 依賴 TaskRepository + EmailSender,最外層由 Celery worker 呼叫 |
| 多租戶 SaaS 平台 | 1. TenantResolver 依賴請求 Header / JWT2. TenantSettings 依賴 TenantResolver(根據租戶讀取不同 DB URL)3. Engine、Session 依賴 TenantSettings,形成「租戶感知」的資料層 |
這些場景的共同點是:每個層級只負責自己的設定或資源,最外層的服務或路由只需要注入最終的「業務物件」,大幅提升程式碼的可讀性與可維護性。
總結
- 多層依賴 是 FastAPI 強大 DI 系統的核心特性之一,讓開發者能把設定、連線、事務、認證、授權等功能以 分層、可重用 的方式組合。
- 正確使用
Depends、use_cache以及 生命週期(scope),可以避免資源浪費與循環依賴的陷阱。 - 透過 工廠函式(如
require_role)與 測試覆寫(dependency_overrides),我們可以在保持程式碼乾淨的同時,快速適應不同的業務需求與測試情境。 - 真正的實務價值在於:路由只需要注入最外層服務,底層的所有依賴都交給 FastAPI 自動解決,讓專案在擴充與維護時更具彈性。
掌握了多層依賴的寫法,你就能在 FastAPI 專案中打造 乾淨、可測、可擴充 的架構,從小型原型一路走向企業級服務。祝開發順利,期待在你的專案裡看到這些技巧的身影! 🚀