FastAPI 教學：依賴注入系統（Dependency Injection） ── Session 管理注入

簡介

在 Web 應用程式中，Session（會話）是用來保存使用者跨請求的狀態資訊，如登入資訊、購物車內容或臨時設定等。若沒有適當的 Session 管理，開發者往往需要在每個路由函式裡手動取得資料庫連線、建立或關閉交易，程式碼會變得冗長且難以維護。

FastAPI 內建的 依賴注入（Dependency Injection, DI） 機制，讓我們可以把 Session 的建立、回收、錯誤處理等流程抽象成可重複使用的函式，然後在需要的地方「注入」進去。這不僅提升程式碼的可讀性，也讓測試變得更簡單，因為我們可以輕鬆替換成 mock 物件。

本篇文章將從核心概念說明起，搭配 3~5 個實用範例，一步步帶你掌握在 FastAPI 中如何正確且有效率地使用 DI 來管理 Session，並討論常見的陷阱與最佳實踐，最後提供實務應用情境，幫助你在真實專案中快速上手。

核心概念

1. 什麼是依賴注入？

依賴注入是一種設計模式，讓函式或類別不直接自行建立所需的資源（例如資料庫連線、Session 物件），而是 由外部提供。在 FastAPI 中，我們透過 Depends 來聲明「我需要這個依賴」，FastAPI 會在請求進來時自動執行相應的依賴函式，並把回傳值當作參數傳入路由函式。

from fastapi import FastAPI, Depends

app = FastAPI()

def get_current_user():
    # 假設這裡會從 token 取得使用者資訊
    return {"username": "alice"}

@app.get("/me")
def read_me(user: dict = Depends(get_current_user)):
    return user

重點：依賴函式本身可以是同步或非同步的，FastAPI 會自行判斷並正確執行。

2. 為什麼要把 Session 當作依賴？

集中管理：所有與 Session 相關的設定（如連線池、事務範圍）集中在一個地方，避免在每個路由裡重複程式碼。
自動釋放：利用 Python 的 yield（生成器）配合 FastAPI 的「依賴生命週期」特性，可在請求結束時自動關閉連線或提交/回滾交易。
測試友好：測試時只需要覆寫 get_db（或 get_session）依賴，即可使用 SQLite 記憶體或 mock 物件，無需改動路由本身。

3. Session 的類型

本文以 SQLAlchemy 2.x 為例，說明兩種常見的 Session 實作方式：

類型	說明	適用情境
同步 Session (`Session`)	傳統的同步 API，適合簡單的服務或已經使用同步框架的程式碼。	小型服務、快速原型
非同步 Session (`AsyncSession`)	基於 `asyncio`，配合 `async`/`await`，可在高併發環境下提升效能。	大型服務、需要同時處理大量 I/O 的 API

以下分別示範兩者的依賴寫法。

4. 程式碼範例

4.1 基礎設定：建立 Engine 與 Base

# db.py
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"   # 只作示範，正式環境請使用 PostgreSQL / MySQL

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)

# 同步 Session
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

註：若要使用非同步，請改用 create_async_engine 與 AsyncSession（見 4.3 範例）。

4.2 同步 Session 的依賴

# dependencies.py
from db import SessionLocal
from fastapi import Depends

def get_db():
    """
    每一次請求都會產生一個獨立的 Session，請求結束後自動關閉。
    使用 `yield` 讓 FastAPI 能在回傳後執行清理程式碼。
    """
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

使用方式：

# main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from dependencies import get_db
from models import User  # 假設已經定義好 User ORM

app = FastAPI()

@app.get("/users/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return {"id": user.id, "name": user.name}

要點：Depends(get_db) 會把 yield 前的 db 物件傳入 read_user，而 finally 區塊則在回應送出後自動執行，確保連線被釋放。

4.3 非同步 Session 的依賴

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

ASYNC_DATABASE_URL = "sqlite+aiosqlite:///./async_test.db"

async_engine: AsyncEngine = create_async_engine(
    ASYNC_DATABASE_URL, echo=True, future=True
)

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

# async_dependencies.py
from fastapi import Depends
from async_db import AsyncSessionLocal

async def get_async_db():
    """
    非同步版的 Session 依賴，使用 async with 讓資源在請求結束時自動釋放。
    """
    async with AsyncSessionLocal() as session:
        yield session

使用方式：

# async_main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from async_dependencies import get_async_db
from models import User  # 同樣的 ORM，只是使用 async 方法

app = FastAPI()

@app.get("/async/users/{user_id}")
async def read_user_async(user_id: int, db: AsyncSession = Depends(get_async_db)):
    result = await db.execute(
        "SELECT id, name FROM users WHERE id = :id", {"id": user_id}
    )
    user = result.first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return {"id": user.id, "name": user.name}

說明：非同步版必須使用 await 觸發資料庫操作，且路由函式本身也要宣告為 async def。

4.4 交易（Transaction）控制的依賴

有時候我們希望在同一個請求內執行多個資料庫操作，且這些操作要麼全部成功，要麼全部失敗。這時可以把 transaction 包裝進依賴：

# transaction.py
from db import SessionLocal
from fastapi import Depends

def get_db_transaction():
    """
    產生一個支援 transaction 的 Session。
    在 yield 前開始 transaction，結束時根據例外自動 commit 或 rollback。
    """
    db = SessionLocal()
    try:
        # 開始 transaction
        db.begin()
        yield db
        # 沒有例外則 commit
        db.commit()
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()

使用方式（在同一請求內完成兩筆寫入）：

@app.post("/transfer")
def transfer_funds(
    from_id: int,
    to_id: int,
    amount: float,
    db: Session = Depends(get_db_transaction)
):
    # 扣款
    db.execute(
        "UPDATE accounts SET balance = balance - :amt WHERE id = :uid",
        {"amt": amount, "uid": from_id}
    )
    # 加款
    db.execute(
        "UPDATE accounts SET balance = balance + :amt WHERE id = :uid",
        {"amt": amount, "uid": to_id}
    )
    return {"status": "success"}

要點：若 any SQL 執行拋出例外，except 區塊會觸發 rollback()，確保資料不會半途而廢。

4.5 測試時的 Session 替換

在單元測試或整合測試中，我們常會使用 SQLite 記憶體資料庫，或是直接 mock Session。只要在測試時 覆寫依賴，就可以不改動原始路由程式碼。

# test_main.py
from fastapi.testclient import TestClient
from main import app
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from db 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():
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db

client = TestClient(app)

def test_read_user():
    # 先插入測試資料
    with TestingSessionLocal() as db:
        db.execute("INSERT INTO users (id, name) VALUES (1, 'Bob')")
        db.commit()
    response = client.get("/users/1")
    assert response.status_code == 200
    assert response.json()["name"] == "Bob"

技巧：app.dependency_overrides 是 FastAPI 提供的測試鉤子，讓你在測試環境輕鬆注入不同的依賴。

常見陷阱與最佳實踐

陷阱	為什麼會發生	解決方案 / 最佳實踐
Session 共享	把全域的 `Session` 直接在模組層級建立，所有請求會共用同一個連線，導致競爭條件與資料不一致。	永遠使用依賴產生的 Session（`yield` 方式），確保每個請求都有獨立的實例。
忘記 `await`	在非同步路由中使用同步 `Session` 或忘記 `await` 呼叫 `execute`，會拋出 `RuntimeWarning` 或阻塞事件迴圈。	使用 AsyncSession 並在所有 I/O 操作前加上 `await`，或在同步路由中使用 `run_in_threadpool`。
Transaction 沒有回滾	只在 `finally` 內關閉 Session，若在 `try` 區塊發生例外，變更仍會被 commit。	在依賴中捕捉例外並呼叫 `rollback()`，或使用 `db.begin()` 產生的 context manager (`async with db.begin():`).
測試環境忘記覆寫依賴	測試時仍連到正式資料庫，造成測試資料污染或意外刪除。	在測試檔案最上方設定 `dependency_overrides`，並使用記憶體資料庫或 mock。
過度依賴全局變數	把 `engine`、`sessionmaker` 直接寫在路由檔案內，導致難以重構與切換 DB。	把 DB 設定抽離成獨立模組（如 `db.py`），並在 `dependencies.py` 中統一管理。

其他最佳實踐

使用 expire_on_commit=False（對於 AsyncSession）以避免在 commit 後自動失效實體，提升後續讀取效能。
將常用的 CRUD 操作封裝成 Repository 類別，再把 Repository 本身作為依賴注入，保持路由乾淨。
在 yield 前先開啟 transaction（db.begin()），在 finally 中僅關閉連線，讓 transaction 的 commit/rollback 完全交給例外處理。
設定適當的連線池大小（pool_size、max_overflow），避免在高併發時產生「Too many connections」錯誤。
記得在 app 關閉時關閉 engine：@app.on_event("shutdown") 中呼叫 engine.dispose()，釋放資源。

實際應用場景

1. 電子商務平台的購物車

使用者在瀏覽商品時會把商品加入購物車，這些資料必須在多個 API 呼叫之間保持一致。透過 DI 注入的 Session，我們可以在每次 add_to_cart、remove_from_cart、checkout 時取得同一個 transaction，確保 購物車變更與訂單建立的原子性。

2. 多租戶 SaaS 系統

每個租戶都有獨立的資料庫或 schema。透過依賴注入，我們可以在 get_tenant_db 中根據 JWT 中的租戶 ID 動態產生對應的 Engine/Session，路由本身不需要關心租戶切換的細節。

def get_tenant_db(tenant_id: str = Depends(get_current_tenant)):
    engine = create_engine(f"postgresql://.../{tenant_id}")
    SessionLocal = sessionmaker(bind=engine)
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

3. 金融交易平台的資金轉移

資金轉移必須在同一個 transaction 中完成，且若任一步驟失敗必須回滾。使用前述 transaction 依賴，可以保證 ACID 特性，同時把錯誤處理集中於依賴層，路由只負責業務驗證。

4. 後台管理系統的批次資料匯入

批次匯入可能涉及上千筆資料寫入。透過 AsyncSession 搭配 async with db.begin():，可以在單一 transaction 中完成大量寫入，減少 I/O 開銷並提升效能。

總結

Session 管理是 Web API 的基礎，不當的連線或交易處理會直接影響系統穩定性與資料正確性。
FastAPI 的 依賴注入機制 提供了一個乾淨、可測試且可擴充的方式來產生與釋放 Session。
透過 yield 搭配 try / finally，我們可以確保每一次請求結束後 自動關閉連線，同時在例外時 自動回滾。
同步 vs. 非同步：根據服務的併發需求選擇 Session 或 AsyncSession，並注意在非同步環境中正確使用 await。
最佳實踐 包括：避免全域 Session、使用 transaction 依賴、在測試中覆寫依賴、適當設定連線池、以及在關閉事件中釋放 engine。

掌握了這套 DI + Session 的設計模式，你就能在 FastAPI 中寫出 乾淨、可靠且易於維護 的資料存取層，為未來的功能擴充與測試奠定堅實基礎。祝你開發順利，Happy Coding! 🚀