FastAPI 與 SQLAlchemy 資料庫整合教學

簡介

在現代的 Web 應用開發中，FastAPI 以其高效、易用且自動產生 OpenAPI 規格的特性，迅速成為 Python 生態系的明星框架。
然而，單純的 API 路由還不足以滿足大部分專案的需求，資料持久化是不可或缺的一環。這時，SQLAlchemy——Python 最成熟、功能最完整的 ORM（Object‑Relational Mapping）工具——就成了最佳拍檔。

本篇文章將從 FastAPI 與 SQLAlchemy 的基本概念說起，示範如何在專案中完成資料庫連線、模型定義、CRUD 操作，並分享常見陷阱與最佳實踐，協助你快速打造可維護、效能良好的後端服務。

---

核心概念

1. 為何選擇 SQLAlchemy？

• 跨資料庫支援：支援 SQLite、PostgreSQL、MySQL、SQL Server 等主流 RDBMS。
• 兩層抽象：提供 Core（SQL 表達式語言）與 ORM（物件導向）兩種使用方式，讓開發者可以依需求自由切換。
• 成熟的事務管理：支援 Session、Transaction、Connection Pool 等高階功能，確保資料一致性。

在 FastAPI 中，我們通常只使用 ORM 方式，因為它與 Pydantic 的資料模型相容性佳，開發體驗更流暢。

2. 建立資料庫引擎與 Session

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

# 1️⃣ 建立資料庫引擎（以 SQLite 為例，實務上多使用 PostgreSQL）
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
# 若使用 PostgreSQL: "postgresql://user:password@localhost/dbname"

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

# 2️⃣ 建立 Session 類別，後續會以此產生 session 物件
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 3️⃣ 建立 Base，用於宣告 ORM Model
Base = declarative_base()

註解

• connect_args 只在 SQLite 時需要，其他資料庫可省略。
• autocommit=False 與 autoflush=False 讓我們能手動控制事務提交，避免不必要的 I/O。

3. 定義 ORM Model

# models.py
from sqlalchemy import Column, Integer, String, Boolean
from .db import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)   # 自動產生 PK
    username = Column(String(50), unique=True, index=True, nullable=False)
    email = Column(String(120), unique=True, index=True, nullable=False)
    is_active = Column(Boolean, default=True)

    def __repr__(self):
        return f"<User(username={self.username!r}, email={self.email!r})>"

重點

• 使用 index=True 能加速搜尋。
• unique=True 讓資料庫自行確保唯一性，減少程式碼裡的驗證成本。

4. 依賴注入（Dependency Injection）取得 Session

FastAPI 的 依賴注入 機制讓我們可以在每個請求中自動產生、關閉資料庫 Session，確保資源不會泄漏。

# dependencies.py
from .db import SessionLocal
from fastapi import Depends, HTTPException, status

def get_db():
    db = SessionLocal()
    try:
        yield db                     # 交給路由使用
    finally:
        db.close()                   # 請求結束後關閉連線

5. CRUD 範例：在路由中使用 ORM

以下示範 建立、讀取、更新、刪除 四個最常見的操作。

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

app = FastAPI()

# 1️⃣ 建立資料表（僅在第一次執行時需要）
models.Base.metadata.create_all(bind=models.engine)

@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)):
    # 檢查使用者名稱或 email 是否已存在
    existing = db.query(models.User).filter(
        (models.User.username == user_in.username) |
        (models.User.email == user_in.email)
    ).first()
    if existing:
        raise HTTPException(status_code=400, detail="Username or email already registered")
    # 建立 ORM 物件後寫入 DB
    user = models.User(**user_in.dict())
    db.add(user)
    db.commit()
    db.refresh(user)   # 取得 DB 產生的 PK
    return user

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

@app.put("/users/{user_id}", response_model=schemas.UserOut)
def update_user(user_id: int, user_in: schemas.UserUpdate,
               db: Session = Depends(dependencies.get_db)):
    user = db.query(models.User).filter(models.User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    for field, value in user_in.dict(exclude_unset=True).items():
        setattr(user, field, value)
    db.commit()
    db.refresh(user)
    return user

@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user(user_id: int, db: Session = Depends(dependencies.get_db)):
    user = db.query(models.User).filter(models.User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    db.delete(user)
    db.commit()
    return

說明

• schemas 是使用 Pydantic 定義的資料驗證模型（稍後會簡述）。
• exclude_unset=True 讓 PUT 時只更新傳入的欄位。
• db.refresh() 可取得剛剛寫入或更新後的最新資料（例如自動產生的 id）。

6. Pydantic 與 SQLAlchemy 的互動

# schemas.py
from pydantic import BaseModel, EmailStr, Field
from typing import Optional

class UserBase(BaseModel):
    username: str = Field(..., max_length=50)
    email: EmailStr

class UserCreate(UserBase):
    password: str = Field(..., min_length=6)   # 實務上會加密儲存

class UserUpdate(BaseModel):
    username: Optional[str] = Field(None, max_length=50)
    email: Optional[EmailStr] = None
    is_active: Optional[bool] = None

class UserOut(UserBase):
    id: int
    is_active: bool

    class Config:
        orm_mode = True   # 讓 Pydantic 能直接從 ORM 物件轉換

重點

• 設定 orm_mode = True 後，Pydantic 會自動讀取 ORM 物件的屬性，省去手動轉換的麻煩。
• 密碼在此示例僅作展示，實務上請使用 bcrypt 或 passlib 進行鹽值雜湊。

---

常見陷阱與最佳實踐

常見問題	為什麼會發生	解決方案 / 最佳實踐
Session 重用	在全域變數中直接建立 Session()，導致多個請求共享同一個連線，產生資料競爭與鎖死。	使用 依賴注入 (Depends(get_db)) 於每個請求建立獨立 Session，請求結束後自動關閉。
忘記 commit()	只呼叫 add() 或 delete() 卻未提交，資料不會寫入資料庫。	操作完畢後一定要 db.commit()，或使用 transaction (with db.begin():) 包裹多個操作。
模型與 Pydantic 不同步	手動維護兩套欄位定義，容易出現欄位遺漏或型別不一致。	盡量使用 SQLModel 或 FastAPI‑SQLAlchemy 之類的工具，把模型與 schema 結合，或在 schemas 中使用 orm_mode。
SQL 注入	手動拼接 SQL 字串，尤其在 Core 層面時容易被注入攻擊。	永遠使用 SQLAlchemy ORM/Expression 提供的參數化查詢 (filter(User.username == username))。
資料庫遷移	直接改動 models.py 而不更新資料庫結構，導致欄位缺失或錯誤。	使用 Alembic 進行版本管理與遷移，搭配 alembic revision --autogenerate 產生變更腳本。

最佳實踐

1. 使用環境變數 管理資料庫連線字串，切勿硬編碼在程式碼。
2. 建立 Connection Pool（engine = create_engine(..., pool_size=10, max_overflow=20)），提升高併發時的效能。
3. 分層架構：把 CRUD 邏輯抽離成 Repository 或 Service 層，讓路由保持簡潔。
4. 測試：使用 pytest 搭配 TestClient 以及 SQLite in‑memory 測試資料庫操作，確保每個 API 都能正確執行。

---

實際應用場景

1. 使用者認證系統

   • 透過 SQLAlchemy 儲存使用者帳號、密碼雜湊、權限角色，結合 FastAPI OAuth2PasswordBearer 完成 JWT 登入流程。

2. 電商商品庫存管理

   • 多表關聯（Product, Category, Inventory）使用 關聯式 (relationship)，在 API 中一次返回商品與庫存資訊，減少前端多次請求。

3. 日誌與審計

   • 利用 SQLAlchemy Events（如 after_insert、after_update）自動寫入操作紀錄表，配合 FastAPI 中間件記錄請求來源 IP、使用者 ID。

4. 多租戶 SaaS 平台

   • 在每筆查詢中注入 tenant_id 條件，或使用 SQLAlchemy 的 schema 功能將不同租戶的資料分離，確保資料隔離性。

---

總結

透過 FastAPI 與 SQLAlchemy 的結合，我們可以在短時間內構建出 高效、結構化且易於維護 的 RESTful API。本文從資料庫引擎、模型定義、依賴注入、CRUD 實作，到常見問題與實務應用，提供了一條完整的開發路徑。

記得：在正式上線前，使用 Alembic 做資料庫遷移管理、加入 單元測試、以及設定 連線池，才能確保系統在大量請求下仍然穩定。

祝你在 FastAPI + SQLAlchemy 的旅程中玩得開心，寫出更好的後端服務！ 🚀