Sharing a database engine¶

A suite that hits a real database needs the connection pool built once, not once per test, while still keeping every test's writes invisible to its neighbours. Two fixtures do this: a session-scoped engine, and a function-scoped session wrapped in a transaction that always rolls back.

@voci.fixture(scope="session")
async def engine(
    tmp: voci.TmpPathFactory = Depends(voci.tmp_path_factory),
) -> AsyncIterator[AsyncEngine]:
    """One engine for the entire run, shared by every concurrent test.

    Session-scoped fixtures are built once, on first use, under a single-flight guard: 200 tests
    starting at the same moment produce exactly one `create_async_engine` call.
    """
    async with database.engine_with_schema(database.url_for(tmp.mktemp("db"))) as e:
        yield e


@voci.fixture()
async def session(engine: AsyncEngine = Depends(engine)) -> AsyncIterator[AsyncSession]:
    """A real session inside a transaction that is always rolled back.

    Every test sees the real schema and none of its neighbours' writes, which is what makes
    hundreds of database tests safe to run at once against a single engine.
    """
    async with database.transaction(engine) as s:
        yield s

scope="session" is what makes engine a run-wide singleton rather than a per-test one; every other fixture scope in voci is per-test. Depending on it from session is enough to reach it — nothing about the dependent fixture needs to know its dependency is shared.

The rollback itself is ordinary SQLAlchemy, not a voci mechanism: session opens a connection, begins a transaction, and hands the test a session bound to it. Whatever the test does — insert, update, even commit(), since the session joins through a SAVEPOINT — is undone by trans.rollback() when the fixture's async with exits, pass or fail.

@asynccontextmanager
async def transaction(engine: AsyncEngine) -> AsyncIterator[AsyncSession]:
    async with engine.connect() as conn:
        trans = await conn.begin()
        async with AsyncSession(
            bind=conn,
            expire_on_commit=False,
            join_transaction_mode="create_savepoint",
        ) as session:
            yield session
        await trans.rollback()

On SQLite specifically, the driver's own autocommit heuristics fight the rollback unless you take over BEGIN yourself — two sqlalchemy.event listeners on connect and on begin, applied once when the engine is built. A suite on Postgres or another server-backed database doesn't need them; they exist here because SQLite is what needs no server to uv sync && voci.

The full fixtures, plus the SQLAlchemy plumbing behind them, are in examples/01-fastapi-crud/tests/fixtures.py and examples/01-fastapi-crud/tests/database.py.