FastAPI – 資料庫整合：SessionLocal 建立

簡介

在使用 FastAPI 開發 Web API 時，最常見的需求之一就是將資料持久化到資料庫。若沒有妥善管理資料庫連線與交易，系統容易出現效能瓶頸、資源洩漏或資料不一致的問題。
SessionLocal（或稱 session factory）是 SQLAlchemy 提供的便利工具，負責在每一次請求期間產生一個 獨立的 Session，讓我們可以安全且有效率地執行 CRUD 操作。

本篇文章將從 概念說明、實作範例、常見陷阱 以及 最佳實踐，一步步帶領讀者完成 SessionLocal 的建立與使用，讓你的 FastAPI 專案在資料庫層面更穩健、更易維護。

---

核心概念

1. 為什麼需要 Session？

• Session 是 SQLAlchemy 與資料庫之間的橋樑，負責追蹤物件狀態、產生 SQL 語句、提交交易等。
• 每一次 HTTP 請求都應該使用 獨立的 Session，避免不同請求之間的資料互相干擾（如未提交的變更、鎖定資源等）。
• 使用 SessionLocal 可以 自動產生 具備正確設定的 Session，並在請求結束時自動關閉，減少資源洩漏的風險。

2. SessionLocal 的設計模式

SessionLocal 本質上是一個 session factory，它是由 sessionmaker 產生的可呼叫物件，內部已經綁定了資料庫引擎（engine）與其他預設參數。典型的建立方式如下：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

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)

• autocommit=False：必須手動呼叫 session.commit()，避免不小心提交未完成的交易。
• autoflush=False：在需要時才寫入資料庫，提升效能。
• bind=engine：將 Session 與先前建立的 engine 連結。

3. 依賴注入（Dependency Injection）與 FastAPI

FastAPI 推崇 依賴注入，讓我們可以把產生 Session 的邏輯抽成一個可重用的函式，並在路由函式中透過 Depends 自動取得 Session：

from fastapi import Depends, FastAPI
from sqlalchemy.orm import Session

app = FastAPI()

def get_db() -> Session:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

• yield 使得 get_db 成為 generator，FastAPI 會在請求結束後自動執行 finally 區塊，關閉 Session。
• 這樣的寫法讓每一個 endpoint 都能得到 乾淨的 Session，而不需要自行管理開關。

4. 與模型（Model）結合

在建立 Session 之後，我們可以使用 Declarative Base 定義模型，並透過 Session 完成 CRUD。以下示範一個簡單的 User 模型與對應的 CRUD 操作。

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id   = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    age  = Column(Integer)

Tip：在專案啟動時，務必先呼叫 Base.metadata.create_all(bind=engine) 以確保資料表已建立。

---

程式碼範例

範例 1：完整的 database.py 設定檔

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

SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost:5432/mydb"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    pool_pre_ping=True,          # 防止連線因閒置被 DB 端斷掉
    pool_size=10,                # 連線池大小，視需求調整
    max_overflow=20              # 超過 pool_size 時的備援連線數量
)

SessionLocal = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=engine,
)

Base = declarative_base()

範例 2：依賴注入取得 Session

# deps.py
from typing import Generator
from sqlalchemy.orm import Session
from .database import SessionLocal

def get_db() -> Generator[Session, None, None]:
    """
    FastAPI 的依賴函式，會在每次請求時產生一個 Session，
    請求結束後自動關閉，確保資源不會泄漏。
    """
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

範例 3：建立 CRUD API（使用 SessionLocal）

# main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from . import models, database, deps

app = FastAPI()

# 初始化資料表
models.Base.metadata.create_all(bind=database.engine)

@app.post("/users/", response_model=dict)
def create_user(name: str, age: int, db: Session = Depends(deps.get_db)):
    """
    新增使用者
    """
    new_user = models.User(name=name, age=age)
    db.add(new_user)
    db.commit()          # 必須手動 commit
    db.refresh(new_user) # 取得寫入後的 primary key
    return {"id": new_user.id, "name": new_user.name, "age": new_user.age}

範例 4：使用 Transaction（交易）保護多筆寫入

@app.post("/users/bulk")
def bulk_create(users: list[dict], db: Session = Depends(deps.get_db)):
    """
    一次寫入多筆資料，若任一筆失敗則全部回滾。
    """
    try:
        for u in users:
            db.add(models.User(**u))
        db.commit()
    except Exception as e:
        db.rollback()   # 回滾所有變更
        raise HTTPException(status_code=400, detail=str(e))
    return {"status": "success", "count": len(users)}

範例 5：自訂 Session 參數（讀寫分離）

# 在 database.py 中再建一個只讀用的 Session
ReadSessionLocal = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=engine,
    future=True,
)

def get_read_db() -> Generator[Session, None, None]:
    db = ReadSessionLocal()
    try:
        yield db
    finally:
        db.close()

在需要大量查詢且不涉及寫入的端點，改用 get_read_db 可以降低寫鎖爭用，提高效能。

---

常見陷阱與最佳實踐

陷阱	說明	解決方案
忘記關閉 Session	若在路由內自行建立 Session，卻未在 finally 中關閉，會導致 DB 連線池耗盡。	使用 依賴注入 + yield 的方式，讓 FastAPI 自動關閉。
在全域變數中直接使用 Session	全域 Session 會在多執行緒環境下共享，同時寫入會產生 race condition。	每次請求產生新的 Session，絕對不要把 SessionLocal() 放在模組層級變數。
自動提交（autocommit=True）	交易自動提交會讓錯誤回滾變得困難，且難以保證資料一致性。	保持 autocommit=False，手動控制 commit() / rollback()。
未設定 pool_pre_ping	長時間閒置的連線可能被資料庫端斷開，導致意外的 OperationalError。	在 create_engine 時加入 pool_pre_ping=True。
使用 SQLite 時忘記 check_same_thread=False	SQLite 預設只允許同一執行緒使用連線，會在多執行緒環境下拋出錯誤。	如範例所示，加入 connect_args={"check_same_thread": False}。

最佳實踐

1. 統一管理 engine、SessionLocal、Base：放在 database.py，其他模組只要 import 即可。
2. 使用 Pydantic Schema 驗證輸入：避免將不合法的資料寫入 DB。
3. Transaction 必要時顯式使用：如多表寫入、批次操作，務必 try/except 包裹 commit/rollback。
4. 設定連線池（pool）參數：根據預期併發量調整 pool_size、max_overflow，避免過度建立連線。
5. 在測試環境使用 SQLite 記憶體資料庫：sqlite:///:memory:，快速且不污染正式資料。

---

實際應用場景

1. 電商平台的訂單系統

   • 每筆訂單需要同時寫入 orders、order_items、inventory 三張表。使用 Transaction 包住所有 Session.add()，若任一筆失敗立即 rollback，確保庫存不會被錯誤扣除。

2. 社群平台的即時訊息

   • 查詢頻繁且寫入較少，使用 只讀 Session (ReadSessionLocal) 處理訊息列表的分頁查詢，減少寫鎖爭用，提升讀取效能。

3. 多租戶 SaaS 系統

   • 每個租戶使用不同的資料庫或 schema，透過 動態建立 engine（根據請求的租戶資訊）再產生對應的 SessionLocal，確保資料隔離。

4. 資料分析後端

   • 大量匯入 CSV/Excel 時，先開啟 bulk insert（session.bulk_save_objects()）或 copy_expert（PostgreSQL），搭配 SessionLocal 進行批次寫入，減少 ORM 產生的額外開銷。

---

總結

• SessionLocal 是 FastAPI 與 SQLAlchemy 整合的核心，負責在每一次請求期間提供 乾淨、獨立且可控的 Session。
• 透過 依賴注入（Depends）與 yield，我們可以自動管理 Session 的生命週期，避免資源洩漏與資料不一致。
• 正確設定 engine、連線池、autocommit/autoflush，並在需要時使用 Transaction，是保證系統穩定性的關鍵。
• 只要遵循本文的 最佳實踐，無論是簡單的 CRUD API，或是複雜的多表交易、讀寫分離，都能在 FastAPI 中以乾淨、可測試的方式完成資料庫整合。

祝開發順利，Happy Coding! 🚀