FastAPI 測試與除錯:測試資料庫
簡介
在開發 FastAPI 應用程式時,資料庫是最常與外部系統互動的關鍵元件。若資料庫操作未經充分測試,就很容易在上線後出現 資料遺失、交易不一致 或 效能瓶頸 等嚴重問題。
本篇文章將說明如何在 FastAPI 專案中建立可靠的資料庫測試環境,從 測試資料庫的建立、事務管理、測試資料的清理 到 常見陷阱與最佳實踐,一步步帶你打造可自動化執行、易於維護的測試流程。文章適合剛接觸 FastAPI 的初學者,也能讓已有經驗的開發者快速回顧測試要點。
核心概念
1. 為什麼要使用獨立的測試資料庫?
- 避免污染正式資料:測試過程中常會執行
INSERT、UPDATE、DELETE,若直接對正式資料庫操作,會造成不可逆的資料變更。 - 可重現性:測試資料庫可以在每次測試前以相同的 schema 與初始資料重新建立,確保測試結果一致。
- 平行測試:CI/CD 流程中會同時跑多個測試工作,獨立的測試資料庫讓每個工作執行環境互不干擾。
2. 測試資料庫的常見類型
| 類型 | 說明 | 優點 | 缺點 |
|---|---|---|---|
| SQLite in‑memory | sqlite:///:memory:,僅在記憶體中運行 |
設定簡單、速度快 | 不支援所有 SQL 功能,與正式 DB 差異較大 |
| Dockerized PostgreSQL/MySQL | 透過 Docker 建立臨時容器 | 與正式環境最相近 | 需要 Docker 環境、啟動較慢 |
| 測試用資料庫 (Test DB) | 於同一 DB 伺服器上建立獨立 schema | 免 Docker、可使用正式 DB 功能 | 需要清理資料、可能受其他測試影響 |
3. 事務 (Transaction) 與測試的關係
在測試中最常見的做法是 在每個測試函式開始時開啟事務,測試結束後回滾,這樣即使測試失敗也不會留下痕跡。FastAPI 與 SQLAlchemy 搭配時,可利用 session.begin_nested() 以及 session.rollback() 來實現。
4. 測試框架與工具
- pytest:最受歡迎的 Python 測試框架,支援 fixture、參數化等功能。
- pytest‑asyncio:提供 async 測試支援,適用於 FastAPI 的非同步端點。
- TestClient:FastAPI 內建的測試客戶端,基於 starlette.testclient,可直接呼叫 API。
程式碼範例
以下範例以 PostgreSQL 為例,展示如何在 FastAPI 中配置測試資料庫、使用 fixture 管理事務、以及撰寫測試案例。所有程式碼均加上說明註解,方便閱讀。
1️⃣ 建立 database.py – 共用的 DB 連線設定
# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
# 正式環境的資料庫 URL
SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost/prod_db"
# 測試環境的資料庫 URL(使用測試專用 schema)
SQLALCHEMY_TEST_URL = "postgresql://user:password@localhost/test_db"
# 依據環境切換
engine = create_engine(
SQLALCHEMY_DATABASE_URL, # 會在測試時被覆寫
pool_pre_ping=True,
)
# Session factory
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
重點:測試時會在
conftest.py中把engine重新指向SQLALCHEMY_TEST_URL,確保不會連到正式資料庫。
2️⃣ 建立 models.py – 簡易的 User 模型
# models.py
from sqlalchemy import Column, Integer, String
from .database import Base
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String, unique=True, index=True, nullable=False)
email = Column(String, unique=True, index=True, nullable=False)
3️⃣ 建立 FastAPI 路由 routers/user.py
# routers/user.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from .. import models, schemas, database
router = APIRouter(prefix="/users", tags=["users"])
def get_db():
db = database.SessionLocal()
try:
yield db
finally:
db.close()
@router.post("/", response_model=schemas.UserOut, status_code=status.HTTP_201_CREATED)
def create_user(user_in: schemas.UserCreate, db: Session = Depends(get_db)):
db_user = db.query(models.User).filter(models.User.email == user_in.email).first()
if db_user:
raise HTTPException(status_code=400, detail="Email already registered")
new_user = models.User(**user_in.dict())
db.add(new_user)
db.commit()
db.refresh(new_user)
return new_user
4️⃣ 測試設定 conftest.py – 使用 pytest fixture 管理測試 DB
# conftest.py
import os
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.main import app # 假設 FastAPI 實例在 main.py
from app import models, database
# -------------------------------------------------
# 1️⃣ 取得測試用資料庫 URL(可從環境變數或 .env 讀取)
TEST_DATABASE_URL = os.getenv(
"TEST_DATABASE_URL",
"postgresql://user:password@localhost/test_db"
)
# 2️⃣ 建立測試用 engine、Session
engine = create_engine(TEST_DATABASE_URL)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# -------------------------------------------------
@pytest.fixture(scope="session")
def db_engine():
"""在測試 session 開始前,建立測試資料庫的 schema。"""
# 建立所有表格(若已存在會自動跳過)
models.Base.metadata.create_all(bind=engine)
yield engine
# 測試結束後,刪除所有表格,保持乾淨
models.Base.metadata.drop_all(bind=engine)
@pytest.fixture(scope="function")
def db_session(db_engine):
"""每個測試函式使用獨立的事務,測試完自動 rollback。"""
connection = db_engine.connect()
transaction = connection.begin()
session = TestingSessionLocal(bind=connection)
# 讓 SQLAlchemy 使用 nested transaction(支援 rollback)
nested = session.begin_nested()
yield session
# 結束測試:回滾事務,關閉連線
session.close()
transaction.rollback()
connection.close()
@pytest.fixture(scope="function")
def client(db_session):
"""為 FastAPI TestClient 注入測試用的 DB session。"""
# 依賴注入:覆寫 get_db 讓它回傳測試用 session
def _get_test_db():
try:
yield db_session
finally:
pass
app.dependency_overrides[database.get_db] = _get_test_db
with TestClient(app) as c:
yield c
# 清除覆寫
app.dependency_overrides.clear()
說明:
db_engine於 session 級別只執行一次,建立/銷毀測試 schema。db_session於 function 級別使用事務,確保每個測試結束後自動 rollback,不留下任何資料。client把 FastAPI 的get_db依賴改寫為測試用 session,讓路由在測試時操作的是測試資料庫。
5️⃣ 撰寫測試案例 tests/test_user.py
# tests/test_user.py
import pytest
from fastapi import status
def test_create_user_success(client):
"""測試正常建立使用者的流程"""
payload = {
"username": "alice",
"email": "alice@example.com"
}
response = client.post("/users/", json=payload)
assert response.status_code == status.HTTP_201_CREATED
data = response.json()
assert data["username"] == "alice"
assert data["email"] == "alice@example.com"
assert "id" in data # 由 DB 自動產生
def test_create_user_duplicate_email(client, db_session):
"""測試重複 email 時會回傳 400 錯誤"""
# 先手動插入一筆資料
from app.models import User
user = User(username="bob", email="bob@example.com")
db_session.add(user)
db_session.commit()
# 再嘗試建立相同 email 的使用者
payload = {
"username": "bob2",
"email": "bob@example.com"
}
response = client.post("/users/", json=payload)
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert response.json()["detail"] == "Email already registered"
@pytest.mark.asyncio
async def test_async_endpoint_example(client):
"""示範如何測試 async 路由(若有)"""
response = await client.get("/async-endpoint")
assert response.status_code == status.HTTP_200_OK
assert response.json()["msg"] == "hello async"
重點:
client直接呼叫 API,測試過程不需要手動建立Session。- 第二個測試示範 先寫入資料 再測試重複檢查,利用
db_session直接操作 DB,保證測試資料在同一事務中。- 使用
pytest.mark.asyncio測試 非同步 路由,展現pytest‑asyncio的配合方式。
6️⃣ 使用 SQLite in‑memory 進行快速測試(可選)
# conftest.py(簡化版,僅用於快速本機測試)
@pytest.fixture(scope="function")
def db_session():
engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False})
models.Base.metadata.create_all(bind=engine)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
session = TestingSessionLocal()
yield session
session.close()
models.Base.metadata.drop_all(bind=engine)
提醒:SQLite 只適合驗證 ORM 的基本行為,若有 Postgres 專屬的功能(如
JSONB、EXCLUDE約束),仍須使用 Dockerized PostgreSQL 進行完整測試。
常見陷阱與最佳實踐
| 陷阱 | 可能的後果 | 解決方案 / 最佳實踐 |
|---|---|---|
| 直接使用正式資料庫 | 資料被測試程式改寫、刪除,甚至造成服務中斷 | 永遠在 conftest.py 中把 engine 換成測試用 URL |
| 測試不回滾事務 | 測試之間相互污染,導致測試結果不一致 | 使用 session.begin_nested() + session.rollback(),或在每個測試結束後手動 drop_all |
| 測試依賴外部服務(如 Redis、第三方 API) | CI 執行失敗、測試速度變慢 | 使用 mock 或 testcontainers 產生臨時容器,僅在需要時才啟動 |
| 忘記清除依賴覆寫 | 之後的測試仍使用測試 DB,或產生奇怪的錯誤 | 在 fixture 結束時 app.dependency_overrides.clear() |
| 使用硬編碼的資料 | 測試不具彈性、難以維護 | 使用 factory_boy 或 pytest‑factoryboy 產生測試資料,保持可讀與可重用性 |
其他最佳實踐
- 測試資料的可預測性:使用 factory 產生唯一的
username、email,避免因唯一鍵衝突導致測試失敗。 - CI 整合:在 GitHub Actions、GitLab CI 中加入
services: postgres,自動啟動測試資料庫容器。 - 測試效能:若測試套件過大,可將 快照測試(snapshot)與 單元測試 分開執行,減少每次 CI 執行時間。
- 文件化測試流程:在
README.md或專案的docs/中說明如何建立本機測試環境,降低新成員的上手門檻。
實際應用場景
| 場景 | 需求 | 方案 |
|---|---|---|
| 新功能上線前的回歸測試 | 確認 CRUD 操作在資料庫層面仍正確 | 使用上述 pytest + TestClient 的全套流程,於 PR 合併前自動跑完整測試 |
| 多租戶 SaaS 系統 | 每個租戶都有獨立 schema,需要在測試中驗證切換正確性 | 在測試 fixture 中動態建立臨時 schema,測試完即刪除 |
| 資料遷移腳本 (Alembic) | 確認 migration 在不同 DB 版本間不會破壞資料 | 在測試中使用 alembic.command.upgrade 與 downgrade,搭配測試 DB 進行驗證 |
| 高併發寫入測試 | 模擬大量同時寫入,檢查 transaction dead‑lock | 使用 asyncio.gather 在 pytest‑asyncio 中同時發送多筆請求,觀察 DB log 是否有 lock 錯誤 |
總結
測試資料庫是 FastAPI 應用程式品質保證的重要環節。透過 獨立的測試資料庫、事務回滾、以及 pytest 的 fixture 結構,我們可以在不影響正式環境的前提下,快速且可靠地驗證所有 CRUD、business logic 與資料遷移的正確性。
本文提供了從 環境配置 → 測試程式碼 → 常見問題 的完整示範,並加入 Docker、SQLite in‑memory、factory 等多種實務技巧,讓讀者能依照專案規模與需求選擇最合適的方案。
記住:測試不只是找錯,更是一種 防止錯誤再次出現 的機制。只要在開發流程中持續執行測試資料庫的自動化檢測,就能大幅降低上線風險,提升團隊對產品的信心。祝開發順利,測試無礙!