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_db→get_current_user→read_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 包裝回傳資料。 |
最佳實踐清單
- 保持函式依賴簡潔:每個依賴只負責一件事(例如「取得 DB Session」或「驗證 Token」)。
- 使用型別註解:讓 IDE 與 FastAPI 都能正確推斷參數與返回值。
- 盡量使用
yield:即使目前不需要清理,也養成寫法,未來擴充時不會遺漏。 - 在測試前覆寫依賴:使用
dependency_overrides,確保測試環境與正式環境徹底分離。 - 避免在依賴內部直接存取 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 等)只寫一次,所有路由自動受惠。
- 確保資源安全:使用
yield或async with,FastAPI 會自動在請求結束時釋放資源。 - 保持程式碼可讀:每個依賴只負責單一職責,讓程式結構清晰、易於維護。
在實務專案中,建議先從 最簡單的函式依賴(如共用參數、簡易驗證)開始,逐步擴展到 層級依賴、類別服務、異步資源,最後根據需求加入 依賴覆寫 以支援測試與環境切換。只要遵循本文的 最佳實踐,你就能在 FastAPI 中建立高品質、易維護且具備彈性伸縮的 API 服務。
祝開發愉快 🎉