如何解决FastAPI异步SQLAlchemy测试中数据库操作冲突问题

问题原因 #

你遇到的InterfaceError本质是测试之间共享了同一个数据库引擎/连接池，且前一个测试的数据库连接未正确释放，导致后续测试的操作与未完成的操作冲突。单独运行测试时，资源可以完全释放，但批量运行时，全局单例引擎的连接池里的连接可能还被前一个测试占用，引发"另一个操作正在进行"的错误。

最优解决方案：用事务回滚替代数据删除 #

相比每次测试删除所有表数据，用事务回滚的方式能更高效地保证测试环境的干净状态，还能避免并发操作冲突。核心思路是：每个测试启动时开启一个独立事务，所有测试操作都在这个事务内执行，测试结束后直接回滚事务，无需删除数据。

修改你的BaseTestCase如下：

from sqlalchemy.ext.asyncio import AsyncSession, sessionmaker

class BaseTestCase(unittest.IsolatedAsyncioTestCase):
    async def asyncSetUp(self) -> None:
        await super().asyncSetUp()
        self.engine = await get_engine()
        
        # 获取独立连接并启动事务
        self.connection = await self.engine.connect()
        await self.connection.begin()
        
        # 创建绑定该连接的异步Session，确保测试中所有DB操作都用这个事务
        self.session_factory = sessionmaker(
            bind=self.connection, 
            class_=AsyncSession, 
            expire_on_commit=False
        )
        self.session = self.session_factory()

        # 替换FastAPI应用中的DB Session依赖（假设你的应用用依赖注入提供Session）
        def override_get_db():
            yield self.session
        my_fastapi_app.dependency_overrides[get_db] = override_get_db

        self.httpclient = AsyncClient(app=my_fastapi_app, base_url="http://test.com")

    async def asyncTearDown(self) -> None:
        # 回滚事务，重置数据库状态
        await self.connection.rollback()
        # 清理资源
        await self.session.close()
        await self.connection.close()
        # 移除依赖覆盖
        my_fastapi_app.dependency_overrides.clear()
        
        await self.httpclient.aclose()
        mockito.verifyStubbedInvocationsAreUsed()
        mockito.unstub()
        await super().asyncTearDown()

备选方案：修复连接池与资源释放 #

如果坚持用删除数据的方式初始化数据库，需调整引擎配置并确保连接正确释放：

1. 调整全局引擎的连接池参数：
   修改get_engine()中的引擎创建代码，限制连接池大小并禁用溢出连接，避免连接堆积：

_engine = create_async_engine(
    db_uri, 
    echo=True,
    pool_size=1,  # 测试环境仅保留1个连接
    max_overflow=0,  # 不允许额外临时连接
    pool_recycle=300  # 定期回收闲置连接，防止失效
)

2. 确保连接正确释放：
   修改asyncSetUp中的数据库清理逻辑，显式获取并关闭连接：

async def asyncSetUp(self) -> None:
    await super().asyncSetUp()
    engine = await get_engine()
    # 显式获取连接，执行清理后确保连接归还给池
    async with engine.connect() as connection:
        async with connection.begin():
            for table in reversed(orm.Base.metadata.sorted_tables):
                await connection.execute(table.delete())
    self.httpclient = AsyncClient(app=my_fastapi_app, base_url="http://test.com")

关键注意事项 #

• 避免测试之间共享数据库连接/事务：每个测试必须使用独立的连接或事务，才能完全隔离状态。
• 事务回滚是测试环境的最优实践：比删除数据更高效，还能避免锁表和操作冲突。
• 检查FastAPI的DB依赖注入：确保测试中所有数据库操作都使用当前测试的事务连接，而不是全局默认的Session。

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