FastAPI内存SQLite结合pytest-xdist并行集成测试报错咨询

核心原因分析 #

你遇到的sqlite3.OperationalError: no such table: certification本质是两个问题叠加：

• 默认的:memory: SQLite数据库是每个连接独立的，主线程初始化表用的是一个连接，但AsyncClient的异步请求在事件循环里会创建新连接，新连接看不到之前创建的表。
• 异步测试场景下，SQLite连接跨线程调用时，默认的check_same_thread=True限制会导致连接异常，间接引发连接重建，进一步加剧表不存在的问题。

具体解决步骤 #

1. 使用共享内存SQLite URL #

放弃:memory:，改用支持多连接共享的内存数据库URL：

• 同步SQLAlchemy：sqlite:///file::memory:?cache=shared
• 异步SQLAlchemy（aiosqlite）：sqlite+aiosqlite:///file::memory:?cache=shared

这个URL会让所有连接指向同一份内存数据库实例，确保表初始化后所有请求都能访问到。

2. 配置连接允许跨线程访问 #

创建SQLite引擎时，添加connect_args={"check_same_thread": False}，解除SQLite对跨线程使用连接的限制（异步测试中请求必然在不同线程执行）。

3. 正确编写Fixture实现依赖注入 #

以下是两种常见场景的Fixture示例：

场景1：同步SQLAlchemy + AsyncClient

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from fastapi.testclient import AsyncClient
from app.main import app
from app.models import Base  # 你的SQLAlchemy基类
from app.dependencies import get_db  # 你的数据库依赖函数

@pytest.fixture(scope="session")
def db_engine():
    # 共享内存SQLite引擎
    engine = create_engine(
        "sqlite:///file::memory:?cache=shared",
        connect_args={"check_same_thread": False}
    )
    # 初始化所有表
    Base.metadata.create_all(bind=engine)
    yield engine
    # 清理
    Base.metadata.drop_all(bind=engine)

@pytest.fixture(scope="function")
def db_session(db_engine):
    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=db_engine)
    session = SessionLocal()
    try:
        yield session
    finally:
        session.close()

@pytest.fixture(scope="function")
async def async_client(db_session):
    # 覆盖数据库依赖，让请求使用当前session
    def override_get_db():
        try:
            yield db_session
        finally:
            pass  # session由fixture统一关闭

    app.dependency_overrides[get_db] = override_get_db
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac
    app.dependency_overrides.clear()

场景2：异步SQLAlchemy + AsyncClient

import pytest
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from fastapi.testclient import AsyncClient
from app.main import app
from app.models import Base
from app.dependencies import get_async_db

@pytest.fixture(scope="session")
async def async_db_engine():
    engine = create_async_engine(
        "sqlite+aiosqlite:///file::memory:?cache=shared",
        connect_args={"check_same_thread": False}
    )
    # 异步创建表
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()

@pytest.fixture(scope="function")
async def async_session(async_db_engine):
    AsyncSessionLocal = sessionmaker(
        async_db_engine, class_=AsyncSession, expire_on_commit=False
    )
    async with AsyncSessionLocal() as session:
        yield session
        await session.rollback()  # 测试后回滚，保证隔离性

@pytest.fixture(scope="function")
async def async_client(async_session):
    def override_get_async_db():
        yield async_session

    app.dependency_overrides[get_async_db] = override_get_async_db
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac
    app.dependency_overrides.clear()

4. pytest-xdist适配注意事项 #

启用xdist后，每个测试进程会独立初始化session级fixture，这意味着每个进程有自己的共享内存数据库，天然实现测试隔离，不需要额外配置。只需确保：

• 表初始化逻辑在session级fixture中，每个进程启动时都会执行
• 不要在进程间共享数据库连接（xdist多进程本身不共享内存，也做不到）

验证要点 #

• 检查所有测试用例是否都通过fixture获取数据库连接，而非直接创建新连接
• 运行单个测试验证表存在问题是否解决，再逐步扩展到批量测试
• 在GitLab CI中测试时，观察IOPS是否下降，超时问题是否缓解

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