FastAPI 教學：依賴注入系統（Dependency Injection）— 資料庫連線注入

簡介

在現代 Web 開發中，資料庫連線是每一個 API 都離不開的基礎設施。若把連線寫死在路由函式裡，程式碼會變得難以測試、難以維護，且在多執行緒或多工作者環境下容易產生資源競爭的問題。

FastAPI 內建的 依賴注入（Dependency Injection, DI） 機制，讓我們能以聲明式的方式把資料庫連線、事務或其他資源「注入」到路由、背景任務或測試函式中。透過 DI，我們可以：

1. 集中管理連線生命週期（建立、關閉、回收）。
2. 輕鬆切換環境（開發、測試、正式），只需要替換依賴實例。
3. 提升可測試性：在單元測試時注入 mock 物件，無需真的連到資料庫。

本篇文章將以 SQLAlchemy 2.x 為例，說明如何在 FastAPI 中實作資料庫連線的依賴注入，從最基本的連線建立到進階的事務管理、異步支援，以及常見的陷阱與最佳實踐。

---

核心概念

1. 為什麼要把「資料庫連線」設為依賴？

在傳統的 Flask 或 Django 中，我們常會在全域變數或 app 物件上直接建立 engine、Session。雖然簡單，但會產生以下問題：

問題	可能的後果
全域單例	多執行緒環境下 Session 可能被多個請求共享，導致資料競爭或不一致。
難以測試	測試時無法輕易替換成測試資料庫或 mock 物件。
資源釋放不當	若忘記在請求結束時關閉連線，會造成連線池耗盡。

FastAPI 的 DI 讓我們把「建立 Session」的行為抽成一個 依賴函式，由框架在每個請求的生命週期自動呼叫，並在請求結束後自動清理。

2. 同步 vs. 非同步的 Session

SQLAlchemy 2.x 同時支援同步 (Session) 與非同步 (AsyncSession) 兩種 API。FastAPI 也支援同步與非同步路由，選擇哪一種取決於：

• 同步：若整個應用大多使用阻塞式程式庫（如 psycopg2），同步 Session 更簡潔。
• 非同步：若想在同一個事件迴圈內同時處理 I/O（例如同時呼叫外部 API、Redis），使用 AsyncSession 可以避免阻塞。

以下範例會同時示範兩種寫法，讓讀者了解差異。

3. 依賴的生命週期（Scope）

FastAPI 允許我們指定依賴的 scope，常見的有：

Scope	說明
request（預設）	每一次 HTTP 請求建立一次，請求結束自動清理。
session	只在測試 TestClient 中使用，與測試會話同生命週期。
singleton	應用啟動時建立一次，常用於不需要關閉的資源（例如配置檔）。

資料庫連線通常使用 request 範圍，確保每個請求都有獨立的 Session。

---

程式碼範例

以下範例使用 Python 3.11、FastAPI 0.110、SQLAlchemy 2.0，資料庫選擇 PostgreSQL（可自行換成 SQLite、MySQL 等）。

3.1 基礎設定：建立 Engine 與 Base

# db.py
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, sessionmaker

# 1️⃣ 建立 Engine（同步版）
SQLALCHEMY_DATABASE_URL = "postgresql+psycopg2://user:password@localhost:5432/mydb"
engine = create_engine(SQLALCHEMY_DATABASE_URL, pool_pre_ping=True)

# 2️⃣ 建立 SessionFactory（同步版）
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 3️⃣ Declarative Base
Base = declarative_base()

註解

• pool_pre_ping=True 能在連線失效時自動重新連線，避免因連線閒置被 DB 端斷開。
• autocommit=False、autoflush=False 為常見的最佳設定，讓我們自行控制事務的提交與刷新。

3.2 同步依賴：在每個請求取得 Session

# dependencies.py
from typing import Generator
from fastapi import Depends, HTTPException
from sqlalchemy.orm import Session
from .db import SessionLocal

def get_db() -> Generator[Session, None, None]:
    """
    FastAPI 依賴函式，負責在請求開始時建立 Session，
    請求結束時自動關閉（即使發生例外）。
    """
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

要點

• 使用 yield 而非 return，FastAPI 會把它視為「依賴的上下文管理器」，在 finally 區塊自動執行清理。
• 若在路由中拋出例外，finally 仍會被呼叫，確保連線不會泄漏。

3.3 非同步依賴：使用 AsyncSession

# db_async.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import declarative_base, sessionmaker

SQLALCHEMY_DATABASE_URL = "postgresql+asyncpg://user:password@localhost:5432/mydb"
async_engine = create_async_engine(SQLALCHEMY_DATABASE_URL, pool_pre_ping=True)

AsyncSessionLocal = sessionmaker(
    bind=async_engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

Base = declarative_base()

# dependencies_async.py
from typing import AsyncGenerator
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from .db_async import AsyncSessionLocal

async def get_async_db() -> AsyncGenerator[AsyncSession, None]:
    """
    非同步版的 DB 依賴，使用 async with 風格的 generator。
    """
    async with AsyncSessionLocal() as session:
        yield session

說明

• expire_on_commit=False 防止在提交事務後自動將物件狀態設為過期，讓返回的 ORM 物件仍可直接使用。
• async with 會在離開區塊時自動呼叫 session.close()。

3.4 路由範例：同步版 CRUD

# main_sync.py
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session
from . import models, schemas, dependencies

app = FastAPI()

@app.post("/users/", response_model=schemas.UserOut, status_code=status.HTTP_201_CREATED)
def create_user(user_in: schemas.UserCreate, db: Session = Depends(dependencies.get_db)):
    """
    建立新使用者的範例。
    1️⃣ 先檢查 email 是否已存在
    2️⃣ 建立 ORM 物件、加入 Session、commit
    """
    existing = db.query(models.User).filter(models.User.email == user_in.email).first()
    if existing:
        raise HTTPException(status_code=400, detail="Email already registered")
    db_user = models.User(**user_in.dict())
    db.add(db_user)
    db.commit()
    db.refresh(db_user)   # 取得自動產生的 ID
    return db_user

重點

• Depends(dependencies.get_db) 讓 FastAPI 把 db 參數自動注入。
• db.refresh() 用來把資料庫回傳的欄位（如主鍵）同步回 ORM 物件。

3.5 路由範例：非同步版 CRUD

# main_async.py
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from . import models, schemas, dependencies_async

app = FastAPI()

@app.post("/items/", response_model=schemas.ItemOut, status_code=status.HTTP_201_CREATED)
async def create_item(item_in: schemas.ItemCreate,
                      db: AsyncSession = Depends(dependencies_async.get_async_db)):
    """
    非同步版的建立 Item，示範 async/await 的寫法。
    """
    # 1️⃣ 檢查是否已存在
    result = await db.execute(
        select(models.Item).where(models.Item.name == item_in.name)
    )
    if result.scalar_one_or_none():
        raise HTTPException(status_code=400, detail="Item already exists")

    # 2️⃣ 建立 ORM 物件並加入 session
    db_item = models.Item(**item_in.dict())
    db.add(db_item)
    await db.commit()
    await db.refresh(db_item)   # 同步回最新狀態
    return db_item

說明

• await db.execute()、await db.commit()、await db.refresh() 必須配合 AsyncSession 使用。
• 若在非同步路由中誤用了同步 Session，會導致整個事件迴圈被阻塞，嚴重影響效能。

3.6 事務（Transaction）管理的進階寫法

有時我們需要在同一個請求中同時操作多個表，且希望所有操作要麼全部成功要麼全部失敗。可以自行寫一個 transaction 依賴：

# dependencies_tx.py
from contextlib import asynccontextmanager
from sqlalchemy.ext.asyncio import AsyncSession
from .db_async import AsyncSessionLocal

@asynccontextmanager
async def get_tx_db() -> AsyncGenerator[AsyncSession, None]:
    """
    取得一個支援事務的 AsyncSession。
    進入 context 時自動 begin()，離開時 commit() 或 rollback()。
    """
    async with AsyncSessionLocal() as session:
        async with session.begin():
            yield session
        # 若程式在此之前拋出例外，session.begin() 內部會自動 rollback

在路由中使用：

@app.post("/orders/")
async def create_order(order: schemas.OrderCreate,
                       db: AsyncSession = Depends(dependencies_tx.get_tx_db)):
    # 多表寫入，若任何一步失敗，全部回滾
    db.add(models.Order(**order.dict()))
    # 可能還要寫入 OrderItem、庫存變更等
    # commit/rollback 交由 get_tx_db 自動處理
    return {"msg": "order created"}

優點

• 只要在 Depends 中使用 get_tx_db，就能保證同一個事務範圍內的所有 DB 操作原子化。
• 透過 asynccontextmanager，程式碼保持簡潔且易於測試。

---

常見陷阱與最佳實踐

陷阱	說明	解決方式
忘記關閉 Session	使用 return 而非 yield，導致 Session 不會在請求結束時關閉。	必須使用 yield，或使用 async with 包裝 AsyncSession。
在非同步路由中使用同步 Session	會阻塞事件迴圈，效能大幅下降。	確認路由是 async def 時，依賴也必須是非同步 (AsyncSession)。
在事務外部手動呼叫 commit()	事務依賴已自動 commit，手動 commit 可能造成二次提交或錯誤。	只在需要自行控制事務時才使用 session.begin()，否則交給 DI 管理。
共用全域 Engine 的 pool size 設定不當	連線池太小會在高併發時耗盡，太大則浪費資源。	依照預期併發量調整 pool_size、max_overflow，或使用 pool_pre_ping 防止失效連線。
測試時未替換資料庫依賴	測試會直接寫入正式資料庫，危險且難以重現。	使用 override_dependency 把 get_db 換成測試用的 SQLite 記憶體或 mock Session。

最佳實踐清單

1. 統一管理 Engine：在單一模組 (db.py / db_async.py) 建立 Engine，其他模組只 import SessionLocal。
2. 使用 yield：所有依賴函式都以 yield 形式提供資源，讓 FastAPI 自動執行清理。
3. 把事務封裝成依賴：如 get_tx_db，讓业务程式碼只關心「做什麼」而不是「怎麼提交」。
4. 環境變數管理：把資料庫 URL、pool 設定放在 .env，使用 pydantic.BaseSettings 讀取。
5. 測試時使用 SQLite 記憶體：sqlite+aiosqlite:///:memory: 讓測試速度極快且不污染正式 DB。
6. 適度使用 expire_on_commit=False：避免在提交後 ORM 物件變成「過期」狀態，導致需要再次查詢。
7. 避免在依賴內做大量計算：依賴應只負責資源取得，業務邏輯放在服務層或路由函式中。

---

實際應用場景

場景	需求	實作方式
多租戶 SaaS	每個租戶使用不同的資料庫 schema，請求開始時根據 JWT 解析租戶 ID，動態切換 Session。	在 get_db 中讀取 tenant_id，使用 sessionmaker(bind=engine_for_tenant) 產生對應 Session。
背景任務（Background Tasks）	異步任務需要存取資料庫，例如發送郵件後更新 email_sent_at。	把 AsyncSession 依賴注入到 background_task 函式，使用 await task(db)。
測試自動化	測試需要在每個測試案例前後重建資料庫結構。	使用 override_dependency 把 get_db 換成 TestingSessionLocal，並在 pytest.fixture 中呼叫 Base.metadata.create_all()。
資料庫讀寫分離	讀取走 Replication，寫入走 Primary。	建立兩個 Engine (engine_read, engine_write)，分別提供 ReadSessionLocal、WriteSessionLocal，在依賴中根據操作類型注入不同 Session。
跨服務交易（Saga Pattern）	多個微服務需要在同一個業務流程中保持資料一致性。	透過 Depends 注入 AsyncSession，在每個服務的 API 呼叫前後使用同一事務（或使用分散式事務協調器）。

---

總結

FastAPI 的依賴注入系統讓 資料庫連線 從「硬編碼」變成「可插拔、可測試」的資源。透過本文的步驟，我們學會：

1. 建立同步與非同步的 Engine、SessionFactory。
2. 使用 yield 方式撰寫 DB 依賴，確保每次請求都會正確關閉連線。
3. 在路由中以 Depends 注入 Session，實作 CRUD 並掌握事務管理。
4. 辨識常見陷阱（忘記關閉、同步/非同步混用）並套用 最佳實踐（環境變數、測試覆寫、事務封裝）。
5. 將 DI 應用於多租戶、背景任務、測試、讀寫分離等實務情境。

掌握了這套「依賴注入 + 資料庫連線」的模式，你的 FastAPI 專案將會變得 更乾淨、更可靠、更易於擴充。未來可以進一步結合 Pydantic v2 的模型驗證、SQLModel 或 Tortoise‑ORM，讓資料層的開發體驗更上一層樓。祝開發順利，寫出高品質的 API！