FastAPI 課程：依賴注入系統（Dependency Injection）

主題：類別依賴注入

---

簡介

在 FastAPI 中，依賴注入（Dependency Injection，簡稱 DI）是實作「可測試、可維護、低耦合」程式碼的關鍵機制。大部分教學會先從函式依賴說起，但在真實專案裡，我們常會把資料庫連線、服務類別（Service）、儲存庫（Repository）等抽象成 類別，再以 DI 方式注入到路由或其他類別中。

使用類別作為依賴有兩大好處：

1. 封裝商業邏輯：將相關方法與狀態集中於同一個物件，讓程式碼更具可讀性。
2. 支援生命週期管理：FastAPI 可以根據請求（request）或應用程式（application）層級，自動建立、釋放物件，避免資源泄漏。

本篇文章將從概念說明、實作範例、常見陷阱、最佳實踐以及實務應用，完整介紹 類別依賴注入 的使用方法，適合剛接觸 FastAPI 的新手，也能讓已有基礎的開發者快速上手。

---

核心概念

1️⃣ 為什麼要用「類別」作為依賴

函式 依賴的寫法簡潔，但在以下情況下會顯得力不從心：

情境	函式依賴的挑戰	類別依賴的優勢
多個 API 需要共用同一套資料庫連線	必須在每個函式內自行建立或傳入連線物件	單例或 scoped 生命週期由 FastAPI 管理，避免重複連線
商業邏輯跨多個方法	函式會彼此依賴，程式碼散落	相關方法封裝在同一個 Service 類別，易於測試與重構
需要在啟動/關閉時執行資源初始化/釋放	必須手動掛鉤事件	類別的 __init__ / __del__ 或 @contextmanager 可自動配合 FastAPI 事件

結論：當依賴涉及「狀態」或「多個行為」時，使用類別是更自然的選擇。

---

2️⃣ 基本寫法：把類別註冊為依賴

FastAPI 只要把類別當成 callable（可呼叫物件）傳入 Depends，框架就會自動執行它的 __call__（如果有實作）或直接建立實例。最簡單的方式是把類別本身交給 Depends：

# example_01.py
from fastapi import FastAPI, Depends

app = FastAPI()

class GreetingService:
    """簡單的問候服務，沒有外部資源"""
    def get_message(self, name: str) -> str:
        return f"哈囉，{name}！歡迎使用 FastAPI。"

def get_greeting_service() -> GreetingService:
    """依賴工廠函式，回傳 GreetingService 實例"""
    return GreetingService()

@app.get("/hello/{name}")
def hello(name: str, service: GreetingService = Depends(get_greeting_service)):
    # 直接使用類別的方法
    return {"message": service.get_message(name)}

• get_greeting_service 為「依賴工廠」，讓我們在未來需要加入初始化參數時，只要修改這個函式即可。
• GreetingService 本身不需要實作 __call__，FastAPI 會直接把工廠函式的回傳值注入。

---

3️⃣ 生命週期與作用域（Scope）

FastAPI 支援三種主要的 作用域（scope）：

Scope	說明	常見使用情境
request（預設）	每一次 HTTP 請求都會建立一次實例	請求內部的資料庫 transaction、使用者驗證服務
session	只在同一個 WebSocket 連線期間保持	WebSocket 訊息處理、長連線緩存
application	整個應用程式啟動期間只建立一次	單例的 Redis 連線池、外部 API 客戶端

以下範例示範 application scope 的類別依賴，適合建立一次性的資源（例如 Redis）：

# example_02.py
import aioredis
from fastapi import FastAPI, Depends

app = FastAPI()

class RedisClient:
    """封裝 aioredis 連線池的類別"""
    def __init__(self, url: str = "redis://localhost"):
        self._url = url
        self._pool = None

    async def connect(self):
        self._pool = await aioredis.from_url(self._url)

    async def get(self, key: str) -> str:
        return await self._pool.get(key, encoding="utf-8")

    async def close(self):
        await self._pool.close()

# 把 RedisClient 設定為 application scope
async def get_redis_client() -> RedisClient:
    client = RedisClient()
    await client.connect()
    return client

# FastAPI 會在應用啟動時建立一次，關閉時釋放
@app.on_event("startup")
async def startup_event():
    app.state.redis = await get_redis_client()

@app.on_event("shutdown")
async def shutdown_event():
    await app.state.redis.close()

def get_redis() -> RedisClient:
    return app.state.redis

@app.get("/cache/{key}")
async def read_cache(key: str, redis: RedisClient = Depends(get_redis)):
    value = await redis.get(key)
    return {"key": key, "value": value}

• RedisClient 只在 startup 時建立一次，之後的每個請求都共用同一個連線池。
• 透過 app.state 把實例掛在 FastAPI 應用上，可在任意路由中透過 Depends(get_redis) 取得。

---

4️⃣ 結合 Pydantic 設定與類別依賴

在大型專案裡，我們往往會把設定寫在 .env、settings.py，並以 Pydantic BaseSettings 讀取。將設定類別作為依賴，可讓服務類別自動取得所需參數：

# example_03.py
import os
from pydantic import BaseSettings
from fastapi import FastAPI, Depends

class Settings(BaseSettings):
    db_url: str = "sqlite:///./test.db"
    api_key: str = "default_key"

    class Config:
        env_file = ".env"

def get_settings() -> Settings:
    """單例設定，使用 application scope"""
    return Settings()

class DatabaseService:
    """使用 Settings 注入資料庫連線字串"""
    def __init__(self, settings: Settings = Depends(get_settings)):
        self._db_url = settings.db_url
        # 這裡假設使用 SQLAlchemy 建立 engine
        # self.engine = create_engine(self._db_url)

    def get_url(self) -> str:
        return self._db_url

app = FastAPI()

@app.get("/db-url")
def show_db_url(service: DatabaseService = Depends()):
    # 直接注入 DatabaseService，FastAPI 會自動解析 Settings
    return {"db_url": service.get_url()}

• Settings 只會在 application 期間建立一次（因為 get_settings 沒有 async，FastAPI 會快取結果）。
• DatabaseService 只要在建構子裡 Depends(get_settings)，就能自動取得設定，且不必在每個路由手動傳遞。

---

5️⃣ 依賴於其他類別（巢狀依賴）

類別依賴可以相互嵌套，形成 巢狀依賴，這是實作「Repository + Service」模式的典型寫法：

# example_04.py
from fastapi import FastAPI, Depends

app = FastAPI()

# 1️⃣ Repository：負責資料存取
class UserRepository:
    def __init__(self):
        # 假設這裡連接資料庫
        self._users = {"alice": {"id": 1, "name": "Alice"}}

    def get_user(self, username: str):
        return self._users.get(username)

# 2️⃣ Service：封裝商業邏輯，依賴 Repository
class UserService:
    def __init__(self, repo: UserRepository = Depends()):
        self._repo = repo

    def fetch_user(self, username: str):
        user = self._repo.get_user(username)
        if not user:
            raise ValueError(f"User {username} not found")
        # 加入額外的商業規則，例如過濾敏感欄位
        return {"id": user["id"], "name": user["name"]}

# 3️⃣ 路由：直接注入 Service
@app.get("/users/{username}")
def read_user(username: str, service: UserService = Depends()):
    try:
        return service.fetch_user(username)
    except ValueError as exc:
        return {"error": str(exc)}

• UserService 只需要 UserRepository，不必自行建立。
• FastAPI 會自動解析 UserRepository 的依賴，形成 依賴樹（dependency graph）。
• 這樣的設計讓 單元測試 非常簡單：只要替換 UserRepository 為 mock 即可。

---

程式碼範例彙總

範例	重點	生命週期
example_01.py	最簡單的類別依賴	request
example_02.py	Application scope（單例）	application
example_03.py	結合 Pydantic Settings	application
example_04.py	巢狀依賴（Repository → Service）	request（預設）
example_05.py	使用 ContextManager 管理交易	request

# example_05.py
from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends

app = FastAPI()

@asynccontextmanager
async def db_transaction():
    # 假設開始 transaction
    print("Transaction START")
    try:
        yield
        # 假設提交 transaction
        print("Transaction COMMIT")
    except Exception:
        # 假設回滾 transaction
        print("Transaction ROLLBACK")
        raise

class TransactionService:
    def __init__(self, ctx = Depends(db_transaction)):
        self._ctx = ctx

    async def do_something(self):
        # 在 transaction 內執行
        async with self._ctx:
            # 這裡寫資料庫操作
            return {"status": "ok"}

@app.post("/process")
async def process(service: TransactionService = Depends()):
    return await service.do_something()

---

常見陷阱與最佳實踐

陷阱	可能的後果	解決方案 / 最佳實踐
忘記設定 scope，導致每次請求都重新建立連線	資源浪費、效能下降	使用 @lru_cache 或 app.state 讓單例在 application 期間只建立一次
在類別 __init__ 中使用 async I/O	__init__ 不能是 async，會拋出 RuntimeError	把 async 初始化搬到 factory 函式 或 @asynccontextmanager 中
巢狀依賴過深，導致循環依賴	FastAPI 報 Dependency cycle detected	重新設計依賴圖，或使用 介面抽象（protocol）與 依賴注入容器（如 injector）
在測試中直接使用實體類別，造成外部資源被呼叫	測試不穩定、速度慢	為每個依賴提供 Mock 或 Stub，利用 override_dependency 替換
在 Depends 內部直接拋出 HTTPException，但未捕捉	例外會直接回傳 500，難以定位問題	在服務層內拋出自訂例外，於路由層捕捉並轉換為適當的 HTTP 狀態碼

最佳實踐

1. 將類別建構與資源初始化分離：使用工廠函式或 @asynccontextmanager，保持 __init__ 同步。
2. 明確宣告作用域：對於資料庫、快取等高成本資源，使用 application scope；對於請求相關的驗證服務，保留預設 request。
3. 利用 @lru_cache（或 functools.cache) 快取設定類別**：避免每次請求都重新讀取 .env。
4. 寫好單元測試：對每個 Service/Repository 建立獨立測試，使用 app.dependency_overrides 注入 mock。
5. 保持依賴樹的深度在 2~3 層：過深的依賴會讓除錯變得困難，盡量把共用邏輯抽到基底類別或工具函式。

---

實際應用場景

場景	為什麼適合類別依賴	範例概念
多租戶 SaaS 平台：每個租戶有不同的資料庫連線	依租戶建立 DatabaseService，在每次請求的 tenant_id 內部動態切換連線。	使用 request scope，並在 Depends 中根據 Header 解析租戶資訊。
第三方支付 API 客戶端：需要統一的金鑰、重試機制	把金鑰、Session、重試策略封裝成 PaymentClient，全站共用同一個實例。	application scope + @lru_cache 快取金鑰設定。
即時聊天（WebSocket）：每條連線需要保持使用者上下文	把 UserContext 類別注入到 WebSocket 路由，確保每個連線都有獨立的使用者資訊。	session scope，與 WebSocket 端點配合。
背景任務（Celery / RQ）：需要共享資料庫與快取	把 Repository 與 CacheService 包成 TaskBase，讓每個 Celery 任務只要繼承即可使用。	application scope，於 worker 啟動時建立。
A/B 測試或功能旗標：根據使用者屬性決定行為	把 FeatureFlagService 包成類別，內部根據設定檔或遠端服務動態決策。	request scope，依賴 UserService 取得使用者屬性。

---

總結

• 類別依賴注入 是 FastAPI 提供的強大機制，讓我們能夠把 狀態、資源、商業邏輯 以物件導向方式封裝，並透過框架自動管理生命週期。
• 了解 作用域（scope）、工廠函式、以及 巢狀依賴 的運作原理，是避免資源浪費與循環依賴的關鍵。
• 結合 Pydantic Settings、ContextManager 與 測試覆寫，可以在開發、部署、測試三個階段都保持程式碼的乾淨與可維護。
• 在實務專案中，從 資料庫連線、快取服務、第三方 API 客戶端 到 WebSocket 使用者上下文，幾乎所有需要共享或有狀態的元件，都可以使用類別依賴來實作。

掌握以上概念與最佳實踐後，你將能夠在 FastAPI 中寫出 乾淨、可測、可擴充 的程式碼，讓專案在面對日益複雜的需求時，仍能保持高效能與低耦合。祝開發順利，期待在你的下一個 FastAPI 專案裡看到優雅的類別依賴注入！