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。 | 依賴抽象化為「可注入」的參數,測試時只要提供不同的實例即可。 |
- 降低耦合度:業務邏輯不再直接建立或管理資源。
- 提升測試便利性:可以在測試環境注入 Fake、Stub 或 In‑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_cache、Depends(..., 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_cache讓get_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 統一管理 |
最佳實踐
- 明確分層:將「外部資源建立」與「業務邏輯」徹底分離,依賴只負責提供資源。
- 使用型別提示:FastAPI 會根據型別自動產生 OpenAPI,讓 IDE 以及文件更友好。
- 最小化依賴樹深度:過深的依賴樹會讓除錯變困難,適度抽象即可。
- 在
startup/shutdown事件中建立長期資源(如模型載入),避免每次請求重新載入。 - 測試時善用
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_db、get_redis 等依賴 |
總結
FastAPI 的依賴注入系統以 宣告式、自動解析 為核心,讓開發者可以把 「取得資源」 的工作交給框架,自己只專注於 業務邏輯。透過 IoC 思維,我們將資源的建立與使用解耦,從而:
- 提升程式可測試性:測試時只要替換依賴即可,不必改動原始程式碼。
- 統一資源管理:生命週期(request、session、singleton)清晰可控,避免資源洩漏。
- 加速開發:共用參數、認證、設定等常見需求只要寫一次依賴函式,即可在多個路由間重用。
掌握了 Depends、yield、@lru_cache 以及 dependency_overrides,你就能在 FastAPI 中自如地運用 IoC,構建出乾淨、可維護、易測試的現代 Web API。祝你開發順利,快把這套思維應用到下一個專案吧!