如何在FastAPI中玩轉“時光倒流”的數據庫事務回滾測試？詳情 - fastapi,django,flask,python,後端風流倜儻的傷痕動態日志

url: /posts/bf9883a75ffa46b523a03b16ec56ce48/
title: 如何在FastAPI中玩轉“時光倒流”的數據庫事務回滾測試？
date: 2025-09-09T04:07:19+08:00
lastmod: 2025-09-09T04:07:19+08:00
author: cmdragon

summary:
在 FastAPI 項目中，集成測試通過事務回滾機制確保測試環境的乾淨性。使用 pytest、SQLAlchemy 和 FastAPI TestClient 組合，實現數據庫事務的嵌套控制，測試中的所有數據庫操作在用例結束時自動回滾。通過 begin_nested() 創建保存點，確保每個測試用例在獨立的事務中執行，避免數據污染。測試案例模擬用户註冊和資料修改，驗證數據庫寫入和接口請求的正確性。常見問題如 IntegrityError 和連接未釋放，通過檢查事務回滾機制和顯式關閉連接解決。

categories:

fastapi

tags:

FastAPI

集成測試
事務回滾
pytest
SQLAlchemy
異步測試
數據庫測試

掃描二維碼關注或者微信搜一搜：編程智域前端至全棧交流與成長

發現1000+提升效率與開發的AI工具和實用程序：https://tools.cmdragon.cn/

1. 多模塊集成測試實踐

在 FastAPI 項目中，隨着功能模塊增多，集成測試成為確保系統整體穩定性的關鍵。

1.1 為何需要事務回滾測試

想象你正在玩一個電子遊戲：每次測試用例就像一局新遊戲。如果上一局的修改沒有撤銷，下一局就會在髒數據上開始，導致結果不可預測。事務回滾測試正是通過 "時光倒流"機制，讓每個測試用例都在乾淨的數據庫環境中運行。

1.2 核心實現方案

使用 pytest + SQLAlchemy + FastAPI TestClient 組合，通過以下組件實現事務控制：

組件	作用	版本要求
pytest	測試框架	>=7.0
SQLAlchemy	ORM 數據庫操作	==2.0.28
FastAPI TestClient	模擬 HTTP 請求	==0.109.0
pytest-asyncio	支持異步測試	==0.23.6

2. 數據庫事務回滾測試模式

通過事務嵌套實現 "沙盒環境"，測試中所有數據庫操作會在用例結束時自動回滾，就像從未發生過。

2.1 實現流程圖

graph TD
    A[測試開始] --> B[創建新事務]
    B --> C[執行測試操作]
    C --> D[接口請求]
    D --> E[數據庫修改]
    E --> F[結果斷言]
    F --> G[事務回滾]
    G --> H[清理資源]

2.2 代碼實現

創建 tests/conftest.py 配置文件：

# 所需依賴：pytest==7.4.4, sqlalchemy==2.0.28, pytest-asyncio==0.23.6
import pytest
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

# 初始化異步數據庫引擎
TEST_DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/test_db"
engine = create_async_engine(TEST_DATABASE_URL, echo=True)

# 配置異步session工廠
AsyncTestingSessionLocal = sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autocommit=False,
    autoflush=False
)

# 核心事務回滾夾具
@pytest.fixture(scope="function")
async def db_session():
    """創建嵌套事務的沙盒環境"""
    async with AsyncTestingSessionLocal() as session:
        async with session.begin_nested():  # 創建SAVEPOINT
            yield session
        # 退出時自動回滾所有操作

3. 完整測試案例

模擬用户註冊+資料修改的集成測試場景。

3.1 測試代碼

# tests/test_user_integration.py
from fastapi.testclient import TestClient
from sqlalchemy import select
from app.main import app
from app.models import User

client = TestClient(app)

@pytest.mark.asyncio
async def test_user_flow(db_session: AsyncSession):
    # 1. 註冊新用户
    res = client.post("/users/", json={"name": "test", "email": "test@demo.com"})
    assert res.status_code == 201
    
    # 2. 驗證數據庫寫入（此時在事務內可見）
    result = await db_session.execute(select(User).where(User.email == "test@demo.com"))
    user = result.scalar_one()
    assert user.name == "test"
    
    # 3. 修改用户資料
    res = client.patch(f"/users/{user.id}", json={"name": "updated"})
    assert res.status_code == 200
    
    # 4. 再次驗證
    await db_session.refresh(user)
    assert user.name == "updated"
    
# 測試結束時，所有數據庫操作自動回滾

3.2 關鍵機制解析

begin_nested()：創建 SQL SAVEPOINT 保存點
yield session：測試代碼在此作用域執行
退出作用域：自動執行 ROLLBACK TO SAVEPOINT
原子性：整個測試過程中即使有多個 DB 操作，也會被視作單一事務單元

4. 常見場景適配方案

4.1 第三方服務模擬

當集成支付等外部服務時，使用 responses 庫攔截 HTTP 請求：

# 新增依賴：responses==0.24.1
import responses

@responses.activate
def test_payment_flow():
    responses.post("https://pay.demo.com", json={"success": True})
    res = client.post("/payments", json={"amount": 100})
    assert res.json()["status"] == "completed"

4.2 異步任務測試

對於 Celery 等異步任務，採用 pytest_celery 插件：

# 新增依賴：pytest-celery==0.0.1
from pytest_celery import CeleryTestWorker

def test_async_task(celery_worker: CeleryTestWorker):
    res = client.post("/tasks/email", json={"to": "user@ex.com"})
    task_id = res.json()["id"]
    
    # 顯式等待任務完成
    result = celery_worker.wait_for_result(task_id)
    assert result["status"] == "sent"

Quiz: 集成測試陷阱排查

❓ 當測試中遇到 IntegrityError 唯一約束衝突時，最可能的原因是什麼？
A) 測試數據清理不徹底
B) 事務隔離級別配置錯誤
C) 未正確啓用回滾機制
D) 數據庫連接池耗盡

<details>
<summary>查看答案</summary>
正確答案：C
解析：事務回滾機制失效會導致測試間數據殘留。應檢查：

是否使用 begin_nested() 創建保存點
yield 後的清理代碼是否被執行
測試數據庫是否配置為可回滾的事務模式
</details>

常見報錯解決方案

報錯 1：`SAVEPOINT does not exist`

現象：

sqlalchemy.exc.InvalidRequestError: This session's transaction has been rolled back 
due to a previous exception during flush...

原因：嵌套事務中執行了 DDL 操作（如 CREATE TABLE）

解決方案：

避免在測試事務中執行遷移操作

初始化時創建好測試表結構：

# 在 conftest.py 中添加
@pytest.fixture(scope="session", autouse=True)
def initialize_db():
 # 在此處運行數據庫遷移腳本

報錯 2：`Connection is not acquired`

現象：

RuntimeError: <Task ...> ... Connection is not acquired.

原因：異步連接未正確釋放

解決方案：

# 改造夾具確保資源釋放
@pytest.fixture
async def db_session():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    async with AsyncTestingSessionLocal() as session:
        try:
            async with session.begin_nested():
                yield session
        finally:
            await session.close()  # 確保顯式關閉連接

餘下文章內容請點擊跳轉至個人博客頁面或者掃碼關注或者微信搜一搜：編程智域前端至全棧交流與成長，閲讀完整的文章：如何在FastAPI中玩轉“時光倒流”的數據庫事務回滾測試？

<details>
<summary>往期文章歸檔</summary>

如何在FastAPI中優雅地模擬多模塊集成測試？ - cmdragon's Blog
多環境配置切換機制能否讓開發與生產無縫銜接？ - cmdragon's Blog
如何在 FastAPI 中巧妙覆蓋依賴注入並攔截第三方服務調用？ - cmdragon's Blog
為什麼你的單元測試需要Mock數據庫才能飛起來？ - cmdragon's Blog
如何在FastAPI中巧妙隔離依賴項，讓單元測試不再頭疼？ - cmdragon's Blog
如何在FastAPI中巧妙隔離依賴項，讓單元測試不再頭疼？ - cmdragon's Blog
測試覆蓋率不夠高？這些技巧讓你的FastAPI測試無懈可擊！ - cmdragon's Blog
為什麼你的FastAPI測試覆蓋率總是低得讓人想哭？ - cmdragon's Blog
如何讓FastAPI測試不再成為你的噩夢？ - cmdragon's Blog
FastAPI測試環境配置的秘訣，你真的掌握了嗎？ - cmdragon's Blog
全鏈路追蹤如何讓FastAPI微服務架構的每個請求都無所遁形？ - cmdragon's Blog
如何在API高併發中玩轉資源隔離與限流策略？ - cmdragon's Blog
任務分片執行模式如何讓你的FastAPI性能飆升？ - cmdragon's Blog
冷熱任務分離：是提升Web性能的終極秘籍還是技術噱頭？ - cmdragon's Blog
如何讓FastAPI在百萬級任務處理中依然遊刃有餘？ - cmdragon's Blog
如何讓FastAPI與消息隊列的聯姻既甜蜜又可靠？ - cmdragon's Blog
如何在FastAPI中巧妙實現延遲隊列，讓任務乖乖等待？ - cmdragon's Blog
FastAPI的死信隊列處理機制：為何你的消息系統需要它？ - cmdragon's Blog
如何讓FastAPI任務系統在失敗時自動告警並自我修復？ - cmdragon's Blog
如何用Prometheus和FastAPI打造任務監控的“火眼金睛”？ - cmdragon's Blog
如何用APScheduler和FastAPI打造永不宕機的分佈式定時任務系統？ - cmdragon's Blog
如何在 FastAPI 中玩轉 APScheduler，讓任務定時自動執行？ - cmdragon's Blog
定時任務系統如何讓你的Web應用自動完成那些煩人的重複工作？ - cmdragon's Blog
Celery任務監控的魔法背後藏着什麼秘密？ - cmdragon's Blog
如何讓Celery任務像VIP客户一樣享受優先待遇？ - cmdragon's Blog
如何讓你的FastAPI Celery Worker在壓力下優雅起舞？ - cmdragon's Blog
FastAPI與Celery的完美邂逅，如何讓異步任務飛起來？ - cmdragon's Blog
FastAPI消息持久化與ACK機制：如何確保你的任務永不迷路？ - cmdragon's Blog
FastAPI的BackgroundTasks如何玩轉生產者-消費者模式？ - cmdragon's Blog
BackgroundTasks 還是 RabbitMQ？你的異步任務到底該選誰？ - cmdragon's Blog
BackgroundTasks與Celery：誰才是異步任務的終極贏家？ - cmdragon's Blog

如何在 FastAPI 中優雅處理後台任務異常並實現智能重試？ - cmdragon's Blog
BackgroundTasks 如何巧妙駕馭多任務併發？ - cmdragon's Blog
如何讓FastAPI後台任務像多米諾骨牌一樣井然有序地執行？ - cmdragon's Blog
FastAPI後台任務：是時候讓你的代碼飛起來了嗎？ - cmdragon's Blog
FastAPI後台任務為何能讓郵件發送如此絲滑？ - cmdragon's Blog
FastAPI的請求-響應週期為何需要後台任務分離？ - cmdragon's Blog
</details>

<details>
<summary>免費好用的熱門在線工具</summary>

歌詞生成工具 - 應用商店 | By cmdragon
網盤資源聚合搜索 - 應用商店 | By cmdragon
ASCII字符畫生成器 - 應用商店 | By cmdragon
JSON Web Tokens 工具 - 應用商店 | By cmdragon
Bcrypt 密碼工具 - 應用商店 | By cmdragon
GIF 合成器 - 應用商店 | By cmdragon
GIF 分解器 - 應用商店 | By cmdragon
文本隱寫術 - 應用商店 | By cmdragon
CMDragon 在線工具 - 高級AI工具箱與開發者套件 | 免費好用的在線工具
應用商店 - 發現1000+提升效率與開發的AI工具和實用程序 | 免費好用的在線工具
CMDragon 更新日誌 - 最新更新、功能與改進 | 免費好用的在線工具
支持我們 - 成為贊助者 | 免費好用的在線工具
AI文本生成圖像 - 應用商店 | 免費好用的在線工具
臨時郵箱 - 應用商店 | 免費好用的在線工具
二維碼解析器 - 應用商店 | 免費好用的在線工具
文本轉思維導圖 - 應用商店 | 免費好用的在線工具
正則表達式可視化工具 - 應用商店 | 免費好用的在線工具
文件隱寫工具 - 應用商店 | 免費好用的在線工具
IPTV 頻道探索器 - 應用商店 | 免費好用的在線工具
快傳 - 應用商店 | 免費好用的在線工具
隨機抽獎工具 - 應用商店 | 免費好用的在線工具
動漫場景查找器 - 應用商店 | 免費好用的在線工具
時間工具箱 - 應用商店 | 免費好用的在線工具
網速測試 - 應用商店 | 免費好用的在線工具
AI 智能摳圖工具 - 應用商店 | 免費好用的在線工具
背景替換工具 - 應用商店 | 免費好用的在線工具
藝術二維碼生成器 - 應用商店 | 免費好用的在線工具
Open Graph 元標籤生成器 - 應用商店 | 免費好用的在線工具
圖像對比工具 - 應用商店 | 免費好用的在線工具
圖片壓縮專業版 - 應用商店 | 免費好用的在線工具
密碼生成器 - 應用商店 | 免費好用的在線工具
SVG優化器 - 應用商店 | 免費好用的在線工具
調色板生成器 - 應用商店 | 免費好用的在線工具
在線節拍器 - 應用商店 | 免費好用的在線工具
IP歸屬地查詢 - 應用商店 | 免費好用的在線工具
CSS網格佈局生成器 - 應用商店 | 免費好用的在線工具
郵箱驗證工具 - 應用商店 | 免費好用的在線工具
書法練習字帖 - 應用商店 | 免費好用的在線工具
金融計算器套件 - 應用商店 | 免費好用的在線工具
中國親戚關係計算器 - 應用商店 | 免費好用的在線工具
Protocol Buffer 工具箱 - 應用商店 | 免費好用的在線工具
IP歸屬地查詢 - 應用商店 | 免費好用的在線工具
圖片無損放大 - 應用商店 | 免費好用的在線工具
文本比較工具 - 應用商店 | 免費好用的在線工具
IP批量查詢工具 - 應用商店 | 免費好用的在線工具
域名查詢工具 - 應用商店 | 免費好用的在線工具
DNS工具箱 - 應用商店 | 免費好用的在線工具
網站圖標生成器 - 應用商店 | 免費好用的在線工具
XML Sitemap

</details>

風流倜儻的傷痕動態日志

@fengliudeshanghen

標簽

後端 (616)

JAVA (556)

數據庫 (310)

Python (289)

MySQL (192)

php (185)

go (184)

redis (93)

postgresql (89)

nosql (49)

elasticsearch (48)

mongodb (41)

動態

如何在FastAPI中玩轉“時光倒流”的數據庫事務回滾測試？ - 動態詳情

1. 多模塊集成測試實踐

1.1 為何需要事務回滾測試

1.2 核心實現方案

2. 數據庫事務回滾測試模式

2.1 實現流程圖

2.2 代碼實現

3. 完整測試案例

3.1 測試代碼

3.2 關鍵機制解析

4. 常見場景適配方案

4.1 第三方服務模擬

4.2 異步任務測試

Quiz: 集成測試陷阱排查

常見報錯解決方案

報錯 1：`SAVEPOINT does not exist`

報錯 2：`Connection is not acquired`

Add a new 評論

風流倜儻的傷痕 動態日志

@fengliudeshanghen

標簽

後端 (616)

JAVA (556)

數據庫 (310)

Python (289)

MySQL (192)

php (185)

go (184)

redis (93)

postgresql (89)

nosql (49)

elasticsearch (48)

mongodb (41)

動態

如何在FastAPI中玩轉“時光倒流”的數據庫事務回滾測試？ - 動態 詳情

1. 多模塊集成測試實踐

1.1 為何需要事務回滾測試

1.2 核心實現方案

2. 數據庫事務回滾測試模式

2.1 實現流程圖

2.2 代碼實現

3. 完整測試案例

3.1 測試代碼

3.2 關鍵機制解析

4. 常見場景適配方案

4.1 第三方服務模擬

4.2 異步任務測試

Quiz: 集成測試陷阱排查

常見報錯解決方案

報錯 1：SAVEPOINT does not exist

報錯 2：Connection is not acquired

Add a new 評論

風流倜儻的傷痕動態日志

如何在FastAPI中玩轉“時光倒流”的數據庫事務回滾測試？ - 動態詳情

報錯 1：`SAVEPOINT does not exist`

報錯 2：`Connection is not acquired`