FastAPI — 資料庫整合（Database Integration）

SQLite / MySQL / PostgreSQL 範例

---

簡介

在現代 Web 開發中，API 與資料庫的整合是不可或缺的核心能力。FastAPI 以其高效能、型別安全與自動產生 OpenAPI 文件的特性，成為許多開發者首選的 Python 框架。無論是小型原型（常用 SQLite）或是正式的企業級服務（MySQL、PostgreSQL），只要掌握正確的整合方式，就能讓資料存取既安全又快速。

本篇文章將從 概念說明、實作範例、常見陷阱與最佳實踐，一步步帶你完成 FastAPI 與三大主流資料庫的結合，並提供 實務應用場景，讓你在開發過程中少走彎路、快速上手。

---

核心概念

1. 為什麼選擇 SQLAlchemy？

• SQLAlchemy 是 Python 生態系最成熟的 ORM（Object‑Relational Mapping）工具，支援同步與非同步（asyncio）兩種模式。
• FastAPI 本身支援 依賴注入（Dependency Injection），配合 SQLAlchemy 可以在每一次請求中安全地取得資料庫會話（Session），避免連線洩漏。

小技巧：在開發階段若只需要簡單的 CRUD，使用 SQLModel（由 FastAPI 團隊開發）會更方便，因為它是基於 SQLAlchemy 並自動產生 Pydantic schema。

2. 同步 vs. 非同步資料庫操作

• 同步：engine = create_engine(...)，適合小型或測試環境，程式碼較簡潔。
• 非同步：engine = create_async_engine(...)，配合 async_session，在高併發情境下能更有效利用 I/O，建議在正式服務（尤其是 MySQL、PostgreSQL）使用。

3. 資料庫 URL 格式

DBMS	URL 範例
SQLite	sqlite:///./test.db（同步） sqlite+aiosqlite:///./test.db（非同步）
MySQL	mysql+aiomysql://user:password@host:3306/dbname
PostgreSQL	postgresql+asyncpg://user:password@host:5432/dbname

注意：+aiomysql、+asyncpg 為非同步驅動，必須先安裝相對應的套件（aiomysql、asyncpg）。

---

程式碼範例

以下示範 三個資料庫（SQLite、MySQL、PostgreSQL）在 FastAPI 中的最小可運行範例，包含模型、依賴注入與 CRUD API。

1️⃣ 基礎設定與模型（共用）

# models.py
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Item(Base):
    """商品資料表"""
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(100), nullable=False, index=True)
    description = Column(String(255), nullable=True)

# schemas.py
from pydantic import BaseModel

class ItemCreate(BaseModel):
    name: str
    description: str | None = None

class ItemRead(ItemCreate):
    id: int

    class Config:
        orm_mode = True

2️⃣ SQLite（同步）範例

# db_sqlite.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from .models import Base

SQLITE_URL = "sqlite:///./sqlite_test.db"
engine = create_engine(SQLITE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 建立資料表（僅執行一次）
Base.metadata.create_all(bind=engine)

# main_sqlite.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from . import db_sqlite, models, schemas

app = FastAPI(title="FastAPI + SQLite 範例")

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

@app.post("/items/", response_model=schemas.ItemRead)
def create_item(item: schemas.ItemCreate, db: Session = Depends(get_db)):
    db_item = models.Item(**item.dict())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

@app.get("/items/{item_id}", response_model=schemas.ItemRead)
def read_item(item_id: int, db: Session = Depends(get_db)):
    db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
    if not db_item:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

說明：

• connect_args={"check_same_thread": False} 為 SQLite 在多執行緒環境下的必要參數。
• 只要執行 uvicorn main_sqlite:app --reload，即可在 http://127.0.0.1:8000/docs 看到自動產生的 API 文件。

3️⃣ MySQL（非同步）範例

# db_mysql.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from .models import Base

MYSQL_URL = "mysql+aiomysql://root:password@localhost:3306/fastapi_demo"
engine = create_async_engine(MYSQL_URL, echo=True, future=True)
AsyncSessionLocal = sessionmaker(
    bind=engine, class_=AsyncSession, expire_on_commit=False
)

async def init_db():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

# main_mysql.py
import asyncio
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from . import db_mysql, models, schemas

app = FastAPI(title="FastAPI + MySQL（非同步）範例")

# 啟動時建立資料表
@app.on_event("startup")
async def on_startup():
    await db_mysql.init_db()

async def get_db() -> AsyncSession:
    async with db_mysql.AsyncSessionLocal() as session:
        yield session

@app.post("/items/", response_model=schemas.ItemRead)
async def create_item(item: schemas.ItemCreate, db: AsyncSession = Depends(get_db)):
    db_item = models.Item(**item.dict())
    db.add(db_item)
    await db.commit()
    await db.refresh(db_item)
    return db_item

@app.get("/items/{item_id}", response_model=schemas.ItemRead)
async def read_item(item_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(
        models.Item.__table__.select().where(models.Item.id == item_id)
    )
    db_item = result.scalar_one_or_none()
    if not db_item:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

提示：

• 使用 await db.refresh(db_item) 讓 ORM 把資料庫最新的值（如自動產生的 id）同步回 Python 物件。
• 若要在開發環境熱重載（--reload），請確保 aiomysql 版本支援 asyncio。

4️⃣ PostgreSQL（非同步）範例

# db_postgres.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from .models import Base

POSTGRES_URL = "postgresql+asyncpg://postgres:password@localhost:5432/fastapi_demo"
engine = create_async_engine(POSTGRES_URL, echo=True, future=True)
AsyncSessionLocal = sessionmaker(
    bind=engine, class_=AsyncSession, expire_on_commit=False
)

async def init_db():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

# main_postgres.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from . import db_postgres, models, schemas

app = FastAPI(title="FastAPI + PostgreSQL（非同步）範例")

@app.on_event("startup")
async def startup():
    await db_postgres.init_db()

async def get_db() -> AsyncSession:
    async with db_postgres.AsyncSessionLocal() as session:
        yield session

@app.post("/items/", response_model=schemas.ItemRead)
async def create_item(item: schemas.ItemCreate, db: AsyncSession = Depends(get_db)):
    db_item = models.Item(**item.dict())
    db.add(db_item)
    await db.commit()
    await db.refresh(db_item)
    return db_item

@app.get("/items/{item_id}", response_model=schemas.ItemRead)
async def read_item(item_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(
        models.Item.__table__.select().where(models.Item.id == item_id)
    )
    db_item = result.scalar_one_or_none()
    if not db_item:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

重點：PostgreSQL 的 asyncpg 驅動在效能上表現最佳，尤其在大量讀寫時，可減少資料庫連線的建立與釋放成本。

---

常見陷阱與最佳實踐

陷阱	可能的結果	解決方案 / 最佳實踐
忘記在 startup 事件中建立資料表	首次請求會拋出「table does not exist」錯誤	在 on_event("startup") 中呼叫 Base.metadata.create_all（同步）或 await conn.run_sync(Base.metadata.create_all)（非同步）
使用同步 engine 搭配 async 路由	會阻塞事件迴圈，導致效能下降	保持一致：同步 engine → 同步路由；非同步 engine → async 路由
未設定 expire_on_commit=False	session.commit() 後 ORM 物件的屬性被「過期」，再次存取會觸發額外查詢	在 sessionmaker 中加入 expire_on_commit=False，尤其在返回剛建立的物件時
SQLite 同時寫入衝突	sqlite3.OperationalError: database is locked	使用 連線池（StaticPool）或改用檔案鎖定模式；正式環境建議換成 MySQL/PostgreSQL
硬編碼 DB URL	部署時需要手動修改程式碼	使用環境變數（os.getenv("DATABASE_URL")）或 .env 檔案配合 python‑dotenv 讀取

其他最佳實踐

1. 分層架構：將 CRUD 操作抽離成 repositories 或 services，讓路由層只負責請求/回應的轉換。
2. 交易（Transaction）管理：對於多表寫入，使用 async with db.begin(): 或 session.begin() 以確保原子性。
3. 資料驗證：利用 Pydantic 的 validator 或 constr 限制欄位長度，避免不合法資料寫入資料庫。
4. 測試：使用 SQLite in‑memory（sqlite:///:memory:）作為單元測試 DB，快速且不污染正式資料庫。

---

實際應用場景

場景	推薦資料庫	為什麼選擇
原型開發 / 本地測試	SQLite	零安裝、檔案即資料庫，快速建立與重置
電商平台的訂單服務	PostgreSQL	支援複雜查詢、事務 (transaction) 與行鎖，適合高併發寫入
即時聊天或遊戲排行榜	MySQL (InnoDB)	良好的寫入效能與水平擴充（讀寫分離）
資料分析平台	PostgreSQL + JSONB	支援半結構化資料與高效的聚合函式
微服務間的共享資料	MySQL 或 PostgreSQL（使用 Docker Compose / Kubernetes）	可透過外部服務發佈/訂閱，且支援多實例的連線池管理

範例：假設你在開發一個「書籍管理系統」的微服務，前端使用 React，後端 FastAPI 需要支援 搜尋、分頁、關鍵字全文檢索。在此情況下，選擇 PostgreSQL 並結合 GIN 索引與 tsvector，即可在 API 中提供高效的全文搜尋功能。

---

總結

• FastAPI + SQLAlchemy 為 Python Web 開發提供了同步與非同步兩條路徑，讓你可以依需求自由切換 SQLite、MySQL、PostgreSQL。
• 透過 依賴注入、資料模型（SQLAlchemy + Pydantic），可以把資料庫操作寫得既安全又易於維護。
• 在實作時別忘了 環境變數管理、啟動時建立資料表、以及 交易與連線池 的設定，這些都是避免常見錯誤的關鍵。
• 最後，根據不同的 業務需求（原型、正式服務、資料分析）選擇合適的資料庫，才能發揮 FastAPI 的高效能與開發速度。

祝你在 FastAPI 與資料庫整合的旅程中，寫出 乾淨、可擴充且效能卓越 的 API！ 🚀