FastAPI 資料庫整合：ORM 模型與 Pydantic Schema

---

簡介

在 FastAPI 中，與資料庫的互動往往會同時使用 ORM（Object‑Relational Mapping）模型 與 Pydantic schema。
ORM 負責把資料表映射成 Python 物件，讓開發者可以以物件操作資料；而 Pydantic schema 則是 FastAPI 用來 驗證、序列化與產生 OpenAPI 文件 的核心工具。

將兩者結合，能讓我們在 資料庫層 與 API 層 各司其職：

1. ORM 保證資料的正確寫入與讀取，同時支援關聯查詢、事務等高階功能。
2. Pydantic schema 把從資料庫取得的 ORM 物件轉換成 乾淨、可驗證的 JSON，並在請求進來時自動檢查資料格式。

本篇文章將從概念說明、實作範例、常見陷阱到最佳實踐，帶你一步步完成 FastAPI + SQLAlchemy + Pydantic 的完整流程，適合剛接觸 FastAPI 的初學者，也能為已有經驗的開發者提供實務參考。

---

核心概念

1. 為什麼要分開 ORM 與 Schema

• 職責分離：ORM 只關心資料庫結構與 CRUD，Schema 只關心資料的驗證與輸出。
• 安全性：直接把 ORM 物件回傳給前端容易洩漏敏感欄位（如 hashed_password），使用 Schema 可自行挑選要公開的欄位。
• 文件生成：FastAPI 會自動根據 Pydantic schema 產生 OpenAPI 文件，若直接回傳 ORM，文件會缺失或不正確。

2. 常見的套件組合

功能	套件	說明
ORM	SQLAlchemy（Core + ORM）	最流行的 Python ORM，支援多種資料庫
非同步支援	SQLModel（基於 SQLAlchemy）	內建 Pydantic 支援，適合簡易專案
Schema	Pydantic（v1/v2）	FastAPI 官方驗證工具
連接池	Databases（可選）	為非同步 SQLAlchemy 提供連接池

以下範例以 SQLAlchemy 1.4+（支援 async）與 Pydantic v2 為例，因為 v2 在型別提示與 model_config 上更直觀。

3. 建立資料庫引擎與 Session

# database.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)

# 建立 async session factory
AsyncSessionLocal = sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,   # 防止 ORM 物件在 commit 後被自動失效
)

小技巧：expire_on_commit=False 能避免在 await db.commit() 後，已載入的 ORM 物件被自動「過期」而必須再次查詢。

4. 定義 ORM 模型

# models.py
from sqlalchemy import Column, Integer, String, Boolean, ForeignKey
from sqlalchemy.orm import relationship, declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, nullable=False, index=True)
    email = Column(String(120), unique=True, nullable=False)
    hashed_password = Column(String(128), nullable=False)
    is_active = Column(Boolean, default=True)

    # 一對多關聯
    posts = relationship("Post", back_populates="owner", cascade="all, delete-orphan")

class Post(Base):
    __tablename__ = "posts"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(200), nullable=False)
    content = Column(String, nullable=False)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="posts")

• relationship 讓我們可以透過 user.posts 直接取得該使用者的所有貼文。
• cascade="all, delete-orphan" 確保刪除使用者時，同時刪除其貼文。

5. 建立 Pydantic Schema

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

class PostBase(BaseModel):
    title: str
    content: str

class PostCreate(PostBase):
    pass   # 建立貼文時不需要額外欄位

class PostRead(PostBase):
    id: int
    owner_id: int

    model_config = ConfigDict(from_attributes=True)   # 讓 ORM 物件自動轉換

class UserBase(BaseModel):
    username: str
    email: EmailStr

class UserCreate(UserBase):
    password: str   # 前端傳入明文密碼，稍後會在路由中雜湊

class UserRead(UserBase):
    id: int
    is_active: bool
    posts: List[PostRead] = []   # 嵌套關聯的貼文

    model_config = ConfigDict(from_attributes=True)

• model_config = ConfigDict(from_attributes=True)（Pydantic v2）告訴 Pydantic 直接從 ORM 物件的屬性讀取資料，省去手動 from_orm 的步驟。
• UserRead.posts 使用 List[PostRead]，讓 API 回傳時自動展開貼文資訊。

6. CRUD 操作：從 ORM 到 Schema

以下示範 非同步 CRUD，並說明如何在路由中把 ORM 物件轉換成 Schema。

# crud.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from models import User, Post
from schemas import UserCreate, UserRead, PostCreate, PostRead
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

async def get_user_by_username(db: AsyncSession, username: str) -> User | None:
    result = await db.execute(select(User).where(User.username == username))
    return result.scalars().first()

async def create_user(db: AsyncSession, payload: UserCreate) -> UserRead:
    hashed_pw = pwd_context.hash(payload.password)
    db_user = User(
        username=payload.username,
        email=payload.email,
        hashed_password=hashed_pw,
    )
    db.add(db_user)
    await db.commit()
    await db.refresh(db_user)          # 取得自動產生的 id
    return UserRead.from_orm(db_user)  # 轉成 Pydantic schema

async def create_post(db: AsyncSession, user_id: int, payload: PostCreate) -> PostRead:
    db_post = Post(**payload.model_dump(), owner_id=user_id)
    db.add(db_post)
    await db.commit()
    await db.refresh(db_post)
    return PostRead.from_orm(db_post)

async def get_user_with_posts(db: AsyncSession, user_id: int) -> UserRead | None:
    result = await db.execute(
        select(User).where(User.id == user_id).options(
            # 立刻載入貼文關聯，避免 N+1 問題
            sqlalchemy.orm.selectinload(User.posts)
        )
    )
    user = result.scalars().first()
    if user:
        return UserRead.from_orm(user)
    return None

說明

• model_dump()（Pydantic v2）是 dict() 的新名稱，回傳純 Python dict。
• selectinload 可一次把關聯的 posts 載入，減少查詢次數。

7. FastAPI 路由整合

# main.py
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from database import AsyncSessionLocal, engine
from models import Base
from crud import create_user, create_post, get_user_with_posts
from schemas import UserCreate, UserRead, PostCreate, PostRead

app = FastAPI(title="FastAPI + SQLAlchemy Demo")

# 建立資料表（開發階段使用，正式環境建議使用 Alembic）
@app.on_event("startup")
async def on_startup():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

# 依賴注入：取得 DB Session
async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

@app.post("/users/", response_model=UserRead, status_code=status.HTTP_201_CREATED)
async def api_create_user(payload: UserCreate, db: AsyncSession = Depends(get_db)):
    existing = await get_user_by_username(db, payload.username)
    if existing:
        raise HTTPException(status_code=400, detail="Username already taken")
    return await create_user(db, payload)

@app.post("/users/{user_id}/posts/", response_model=PostRead)
async def api_create_post(user_id: int, payload: PostCreate, db: AsyncSession = Depends(get_db)):
    # 省略權限驗證，實務上應檢查 token 與 user_id 是否相符
    return await create_post(db, user_id, payload)

@app.get("/users/{user_id}", response_model=UserRead)
async def api_get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    user = await get_user_with_posts(db, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

• 透過 response_model，FastAPI 自動把 UserRead、PostRead 轉成 JSON，且會根據 schema 產生 OpenAPI 文件。
• Depends(get_db) 為每次請求提供獨立的 Session，確保連線安全。

---

常見陷阱與最佳實踐

陷阱	為什麼會發生	解決方式
1. 直接回傳 ORM 物件	會把所有欄位（包括敏感資訊）序列化，且 OpenAPI 無法正確描述	永遠使用 Pydantic schema 作為 response_model
2. N+1 查詢問題	迭代關聯集合時，每筆資料都會額外觸發一次 SELECT	使用 selectinload、joinedload 或手動 await db.execute(select(...).options(...))
3. Session 共享	在全域建立 Session，多請求同時使用會導致競爭條件	依賴注入 AsyncSessionLocal()，每次請求建立/關閉
4. 密碼未加鹽或雜湊	明文儲存會造成安全漏洞	使用 passlib、bcrypt 之類的雜湊函式，並在 create_user 時處理
5. expire_on_commit=True	commit 後 ORM 物件屬性會自動失效，需要再次查詢	設為 expire_on_commit=False，或在需要時手動 await db.refresh(obj)

最佳實踐

1. 分層設計：

   • models.py 只放 ORM 定義。
   • schemas.py 只放 Pydantic 定義。
   • crud.py 實作資料庫操作。
   • router/*.py 處理 HTTP 請求與回應。

2. 使用 Alembic 進行 Migration：

   • alembic revision --autogenerate -m "init"
   • 版本管理能避免手動改表造成的同步問題。

3. 驗證輸入資料：

   • 利用 Pydantic 的 EmailStr、constr(min_length=8) 等限制。
   • 在路由層加入 HTTPException，返回統一的錯誤格式。

4. 非同步優化：

   • 若使用 PostgreSQL，建議搭配 asyncpg 驅動。
   • 在大量寫入時，可使用 session.bulk_save_objects() 減少 round‑trip。

5. 測試：

   • 使用 TestClient 搭配 pytest-asyncio 測試非同步路由。
   • 為每個 CRUD 函式寫單元測試，確保資料一致性。

---

實際應用場景

場景	為何需要 ORM + Schema	可能的擴充功能
使用者認證系統	需要儲存密碼雜湊、驗證登入資料；Schema 保護 hashed_password 不被外洩	JWT、OAuth2、Refresh Token
部落格平台	一對多的 User ↔ Post 關聯；Schema 可一次返回使用者資訊與其貼文	分頁、全文搜尋（ElasticSearch）
電商訂單	多表（User、Product、Order、OrderItem）關聯；Schema 讓前端一次取得訂單摘要與明細	支付整合、訂單狀態工作流
即時聊天	訊息儲存於資料庫，同時需要返回訊息的作者資訊；Schema 控制哪些欄位可被外部看到	WebSocket、訊息推播
資料分析儀表板	大量讀取統計資料，使用 ORM 的聚合函式；Schema 定義圖表所需的欄位結構	Pandas、Plotly、Cache（Redis）

以上皆展示了 資料庫層的複雜度 與 API 層的簡潔需求，透過 ORM + Pydantic 的配合，我們可以在保持程式可讀性與安全性的同時，快速產出符合 OpenAPI 規範的服務。

---

總結

• ORM 負責把資料庫的表格映射成 Python 類別，提供 CRUD、關聯與交易管理。
• Pydantic schema 為 FastAPI 的資料驗證與序列化核心，讓 API 輸入/輸出保持一致、文件自動生成。
• 兩者 分工明確、相輔相成：在實作時，先使用 ORM 操作資料，再以 Schema 把結果送回前端。
• 常見的陷阱包括直接回傳 ORM、N+1 查詢、Session 共享與密碼未加密等；透過依賴注入、selectinload、expire_on_commit=False 以及 passlib 等工具即可避免。
• 在真實專案中，搭配 Alembic 做 migration、JWT 做認證、Cache 提升讀取效能，能讓 FastAPI 應用更具擴充性與可維護性。

掌握了 ORM 模型與 Pydantic schema 的正確使用方式，你就能在 FastAPI 中建立既安全又高效的資料庫整合層，為後續的微服務、資料分析或前端互動奠定堅實基礎。祝你開發順利，寫出乾淨、易維護的 API！