FastAPI异步MySQL环境下实现Pytest用例事务隔离自动回滚

我有一个基于FastAPI、MySQL和asyncio的异步应用，希望为其集成测试用例，实现每个测试用例执行后自动回滚数据库变更，保证用例间的隔离性。

现有控制器配置（注入数据库连接依赖） #

from sqlalchemy.ext.asyncio import create_async_engine

async def get_db_connection_dependency():
    engine = create_async_engine("mysql+aiomysql://root:root@mysql8:3306/user_db")
    connection = engine.connect()
    return connection

class UserController:
   async def create_user(
            self,
            request: Request,
            connection: AsyncConnection = Depends(get_db_connection_dependency)

    ) -> JSONResponse:
        
        # START TRANSACTION
        await connection.__aenter__()
        transaction = connection.begin()
        await transaction.__aenter__()

        try:
            do_stuff()
        except:
            await transaction.rollback()
        else:
            await transaction.commit()
        finally:
            await connection.close()
      
        # END TRANSACTION
        
        return JSONResponse(status_code=201)

现有Pytest测试用例 #

import pytest

app = FastAPI()

@pytest.fixture()
def client():
    with TestClient(app=app) as c:
        yield c

class TestUserCreation:
    CREATE_USER_URL = "/users/create"
    
    def test_create_user(self, client):
        response = client.post(self.CREATE_USER_URL, json={"name": "John"})
        assert response.status_code == 201

当前测试用例可正常运行，但会将新创建的用户持久化到数据库中，我需要实现测试结束后自动回滚变更。

我尝试过的方案都不适用：

• 使用factory objects：但我的控制器直接依赖数据库连接并操作数据库，无法使用该方案；
• 手动注入依赖：想在测试前创建连接注入，测试后回滚，但没找到可行的注入方式；
• FastAPI官方文档对数据库测试回滚的介绍不够详尽。

我想到过通过中间件附加连接到请求对象，根据环境变量在测试环境强制回滚，但感觉过度设计。

请问FastAPI是否有内置或现成的库实现该需求？若Pytest不适合，也可更换测试框架。

核心思路是在测试时替换数据库连接依赖，为每个测试用例创建一个带事务的数据库连接，测试结束后自动回滚事务，无需修改业务代码，也不需要过度设计中间件。以下是具体实现步骤：

1. 重构数据库依赖，提升可测试性 #

首先把引擎创建逻辑抽离，避免重复创建引擎，同时简化依赖的上下文管理：

# database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncConnection

# 全局单例引擎（生产/测试环境复用）
engine = create_async_engine("mysql+aiomysql://root:root@mysql8:3306/user_db")

async def get_db_connection_dependency():
    async with engine.connect() as connection:
        yield connection

同时简化控制器的事务逻辑，利用SQLAlchemy上下文管理器自动处理事务生命周期：

class UserController:
   async def create_user(
            self,
            request: Request,
            connection: AsyncConnection = Depends(get_db_connection_dependency)

    ) -> JSONResponse:
        try:
            async with connection.begin() as transaction:
                # 执行实际数据库操作，例如插入用户
                await connection.execute(insert_stmt, {"name": (await request.json())["name"]})
                # 无异常时自动提交事务
        except Exception as e:
            # 异常时自动回滚
            await transaction.rollback()
            raise e
        
        return JSONResponse(status_code=201)

2. 编写测试用例的数据库Fixture #

使用pytest-asyncio创建异步Fixture，为每个测试用例生成独立的事务连接，并替换原有的依赖：

import pytest
from sqlalchemy.ext.asyncio import AsyncConnection
from fastapi.testclient import AsyncTestClient
from your_app import app, engine, get_db_connection_dependency

# 异步测试客户端Fixture
@pytest.fixture(scope="function")
async def async_client():
    async with AsyncTestClient(app=app) as client:
        yield client

# 带事务的数据库连接Fixture，每个测试用例独立
@pytest.fixture(scope="function")
async def test_db_connection():
    async with engine.connect() as connection:
        # 开启事务
        await connection.begin()
        yield connection
        # 测试结束后强制回滚，不提交到数据库
        await connection.rollback()

# 自动替换FastAPI的数据库依赖
@pytest.fixture(autouse=True)
async def override_db_dependency(test_db_connection: AsyncConnection):
    async def override_get_db():
        yield test_db_connection
    app.dependency_overrides[get_db_connection_dependency] = override_get_db
    yield
    # 测试完成后清除依赖覆盖
    app.dependency_overrides.clear()

3. 编写异步测试用例 #

适配异步接口，使用async def测试函数：

class TestUserCreation:
    CREATE_USER_URL = "/users/create"
    
    async def test_create_user(self, async_client):
        response = await async_client.post(self.CREATE_USER_URL, json={"name": "John"})
        assert response.status_code == 201
        # 可额外查询数据库验证操作生效（测试结束后会被回滚）

关键说明 #

• 依赖覆盖：FastAPI的app.dependency_overrides是官方推荐的测试依赖替换方式，无需修改业务代码；
• 事务隔离：每个测试用例使用独立事务，测试结束后回滚，完全保证用例间的数据隔离；
• 异步适配：使用AsyncTestClient替代同步TestClient，适配异步接口测试；
• 无额外依赖：仅使用FastAPI、SQLAlchemy和pytest-asyncio，无需第三方工具库。

内容的提问来源于stack exchange，提问作者Sidharth Samant