FastAPI 資料庫連線注入

單元：資料庫整合（Database Integration）
主題：資料庫連線注入

---

簡介

在建置 API 時，資料庫連線是最常見且最重要的資源。如果每一個路由都自行建立、關閉連線，不僅程式碼會變得雜亂，亦會造成連線資源浪費、效能下降，甚至產生資料不一致的問題。FastAPI 以 依賴注入（Dependency Injection, DI） 為核心設計，讓我們可以在一個統一的地方管理資料庫連線，並在需要的地方自動取得。

透過 DI，我們不僅可以：

1. 統一管理連線生命週期（建立、關閉、回收）。
2. 簡化測試：只要替換依賴，就能在單元測試中使用記憶體資料庫或 mock 物件。
3. 提升可讀性與維護性：路由函式只關注業務邏輯，資料庫細節被抽象到獨立模組。

本篇將從概念說明、實作範例、常見陷阱與最佳實踐，最後延伸到實務應用場景，幫助你在 FastAPI 專案中正確且高效地使用資料庫連線注入。

---

核心概念

1. 為什麼要使用依賴注入管理資料庫連線？

FastAPI 的 Depends 機制允許我們把「取得資料庫 Session」的邏輯寫成一個 可重用的函式，然後在路由中以參數的方式注入。這樣做的好處包括：

• 單一入口：所有連線的建立與關閉都集中在同一個函式，避免遺漏 session.close()。
• 自動回收：FastAPI 會在請求結束時自動呼叫 yield 之後的程式碼，確保資源釋放。
• 支援同步與非同步：依賴函式本身可以是同步或 async，FastAPI 會根據需求調度。

註：在同步環境下使用 SQLAlchemy 的 Session，在非同步環境下則建議使用 SQLModel 或 SQLAlchemy 2.0 的 AsyncSession。

2. 基本的依賴函式範例（同步）

下面示範最簡單的 同步 依賴函式，使用傳統的 SQLAlchemy sessionmaker。

# 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}
)
# 建立 Session 類別
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db() -> Session:
    """
    依賴函式：在每次請求開始時建立 Session，
    請求結束後自動關閉（使用 try/finally）。
    """
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

重點：yield 前建立連線，finally 區塊負責關閉連線，FastAPI 會在請求結束時執行 finally。

3. 非同步的依賴函式（AsyncSession）

在需要高併發的環境，建議改用 AsyncSession。以下範例使用 SQLAlchemy 2.0 的非同步 API。

# async_db.py
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql+asyncpg://user:password@localhost:5432/mydb"

engine: AsyncEngine = create_async_engine(DATABASE_URL, echo=True)

# 建立非同步 Session 類別
AsyncSessionLocal = sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autocommit=False,
    autoflush=False,
)

async def get_async_db() -> AsyncSession:
    """
    非同步依賴函式，使用 async with 讓 Session 在請求結束時自動釋放。
    """
    async with AsyncSessionLocal() as session:
        yield session

提醒：使用 async with 可以確保即使在例外情況下，連線也會正確關閉。

4. 在路由中注入資料庫 Session

# main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from db import get_db
from models import User, UserCreate  # 假設已定義 Pydantic 與 ORM

app = FastAPI()

@app.post("/users/", response_model=User)
def create_user(user_in: UserCreate, db: Session = Depends(get_db)):
    """
    透過 Depends 注入 db Session，接著執行 CRUD。
    """
    db_user = db.query(User).filter(User.email == user_in.email).first()
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    new_user = User(**user_in.dict())
    db.add(new_user)
    db.commit()
    db.refresh(new_user)   # 取得自動產生的 ID
    return new_user

5. 多個依賴函式的組合

有時候我們會同時需要 資料庫 Session 與 認證資訊，只要把多個 Depends 放在同一個參數列表即可。

from fastapi import Security
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def get_current_user(token: str = Security(oauth2_scheme)):
    # 這裡驗證 token，回傳使用者物件
    ...

@app.get("/items/")
def read_items(
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user)
):
    # 兩個依賴同時注入
    items = db.query(Item).filter(Item.owner_id == current_user.id).all()
    return items

6. 在測試環境中覆寫依賴

測試時不想連到正式資料庫，只要 override FastAPI 的依賴即可。

# test_main.py
from fastapi.testclient import TestClient
from main import app
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Base

SQLITE_MEMORY_URL = "sqlite:///:memory:"
engine = create_engine(SQLITE_MEMORY_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 建立測試用的資料表
Base.metadata.create_all(bind=engine)

def override_get_db():
    try:
        db = TestingSessionLocal()
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db

client = TestClient(app)

def test_create_user():
    response = client.post("/users/", json={"email":"test@example.com","name":"Test"})
    assert response.status_code == 200
    assert response.json()["email"] == "test@example.com"

關鍵：app.dependency_overrides 只在測試過程中生效，正式環境不受影響。

---

常見陷阱與最佳實踐

陷阱	說明	解決方式
忘記關閉 Session	若依賴函式沒有使用 yield 或 async with，請求結束後連線不會釋放，導致資料庫連線池耗盡。	必須使用 yield 搭配 try/finally（同步）或 async with（非同步）。
在全域變數中直接建立 Session	直接在模組層級 db = SessionLocal()，所有請求共用同一個 Session，會產生競爭條件與資料不一致。	依賴函式每次請求產生新的 Session。
混用同步與非同步 Session	在同一個路由內同時使用同步 Session 與非同步 AsyncSession，會導致事件迴圈錯誤。	依路由的 async/sync 屬性統一使用相同類型的 Session。
未在測試中覆寫依賴	測試時仍連到實體資料庫，可能污染資料或因環境差異失敗。	使用 app.dependency_overrides 注入測試用資料庫。
大量資料寫入時忘記 commit	db.add() 之後未呼叫 db.commit()，資料不會持久化。	確認每個寫入流程都有 commit（或 await session.commit()）。

最佳實踐清單

1. 統一管理 Engine：在單一模組（如 db.py）建立 engine，避免在多個檔案重複呼叫 create_engine。
2. 使用 Base.metadata.create_all：在開發階段於 startup 事件自動建立資料表，正式環境建議使用 Migration 工具（Alembic）。
3. 限制 Session 的作用範圍：只在需要的路由或服務層級使用 Depends(get_db)，不要在全域變數或類別屬性中保存 Session。
4. 明確設定 autocommit=False、autoflush=False：避免隱式提交或自動刷新造成效能問題。
5. 在非同步環境使用 async with：確保例外時仍能正確釋放連線。
6. 將資料庫操作封裝成 Repository / Service：讓路由只負責參數驗證與回傳，真正的 CRUD 放在獨立的函式或類別中，方便測試與重用。

---

實際應用場景

1. 多租戶（SaaS）系統的動態資料庫切換

在 SaaS 平台，每個租戶可能擁有獨立的資料庫。透過依賴注入，我們可以在每一次請求的 get_db 中根據租戶 ID 動態建立 SessionLocal。

def get_tenant_db(tenant_id: str = Depends(get_current_tenant)):
    db_url = f"postgresql+asyncpg://user:pass@host/{tenant_id}_db"
    async_engine = create_async_engine(db_url, echo=False)
    async_session = sessionmaker(
        bind=async_engine,
        class_=AsyncSession,
        expire_on_commit=False,
    )
    async with async_session() as session:
        yield session

2. 背景工作（Celery / RQ）與 API 共用同一套 Session 工廠

背景任務需要存取資料庫，卻不在 HTTP 請求上下文中。只要直接呼叫 SessionLocal()（或 AsyncSessionLocal()）即可，前提是 使用相同的 Engine，確保連線池一致。

# tasks.py
from db import SessionLocal

def generate_report():
    db = SessionLocal()
    try:
        # 執行大量查詢與寫入
        ...
        db.commit()
    finally:
        db.close()

3. 讀寫分離（Read Replicas）

在高流量網站，寫入使用主庫，查詢使用只讀複本。可以寫兩個依賴函式 get_write_db、get_read_db，在路由上選擇性注入。

def get_write_db() -> Session:
    db = WriteSessionLocal()
    try:
        yield db
    finally:
        db.close()

def get_read_db() -> Session:
    db = ReadSessionLocal()
    try:
        yield db
    finally:
        db.close()

---

總結

• 依賴注入是 FastAPI 管理資料庫連線的核心機制，讓我們能在請求開始時建立 Session，請求結束時自動釋放。
• 同步與非同步兩種實作方式（Session vs AsyncSession）各有適用情境，依需求選擇即可。
• 正確的寫法 必須使用 yield（同步）或 async with（非同步）搭配 try/finally，避免連線洩漏。
• 測試時覆寫依賴 能讓我們在不接觸正式資料庫的情況下完成單元測試，提高開發效率與安全性。
• 最佳實踐 包括統一 Engine、限制 Session 範圍、使用 Repository/Service 層、以及在高階場景（多租戶、讀寫分離）中靈活切換資料庫。

掌握了 資料庫連線注入，你就能在 FastAPI 專案中寫出 乾淨、易測、可擴充 的 API，從而在實務開發中快速迭代、穩定上線。祝開發順利，期待你用 FastAPI 打造更棒的服務！