Rolling back database state between async tests

Bind the test session to a connection whose outer transaction is never committed, pass join_transaction_mode="create_savepoint" to the sessionmaker, and roll that transaction back at teardown — every test then starts from an identical database without truncating a table. This is the fixture at the centre of testing async SQLAlchemy applications.

Quick Answer

Before — truncation between tests:

@pytest_asyncio.fixture(autouse=True)
async def clean_tables(engine):
    yield
    async with engine.begin() as connection:
        for table in reversed(Base.metadata.sorted_tables):
            await connection.execute(text(f'TRUNCATE TABLE "{table.name}" CASCADE'))

After — a never-committed outer transaction:

from collections.abc import AsyncIterator

import pytest_asyncio
from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine, AsyncSession, async_sessionmaker


@pytest_asyncio.fixture
async def connection(engine: AsyncEngine) -> AsyncIterator[AsyncConnection]:
    async with engine.connect() as connection:
        transaction = await connection.begin()
        yield connection
        await transaction.rollback()


@pytest_asyncio.fixture
async def session(connection: AsyncConnection) -> AsyncIterator[AsyncSession]:
    factory = async_sessionmaker(
        bind=connection,                      # bound to the connection, NOT the engine
        expire_on_commit=False,
        join_transaction_mode="create_savepoint",
    )
    async with factory() as session:
        yield session

Two details carry the whole pattern. bind=connection is what puts the session inside the test's transaction rather than drawing a fresh connection from the pool. And join_transaction_mode="create_savepoint" is what lets the code under test commit without committing the outer transaction.

Execution Context & Async Workflow Integration

The fixture works because a connection is a single, ordered conversation with the database. The outer transaction is opened on that conversation; the session joins the same conversation rather than starting a new one; and the rollback ends it. Nothing about that reasoning is async-specific — but two things about asyncio make it easier to get wrong.

The first is loop affinity. The connection belongs to the event loop that created it, so a session-scoped engine handing connections to function-scoped tests works only if the loop is also session-scoped. Configure it once:

[tool.pytest.ini_options]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "session"

The second is that AsyncSession cannot be shared between concurrent tasks. A test exercising concurrency must give each task its own session — and each of those sessions must be bound to its own connection, which means those tasks are not inside the test transaction and their writes will really commit.

# Async — a concurrency test cannot use the rollback fixture for the racing tasks
import asyncio

import pytest


@pytest.mark.asyncio
@pytest.mark.postgres
async def test_concurrent_reservations_conflict(engine, clean_database):
    """Two tasks, two sessions, two real transactions — outside the rollback fixture."""
    factory = async_sessionmaker(engine, expire_on_commit=False)

    async def reserve(seat: int) -> bool:
        async with factory() as session, session.begin():
            return await try_reserve(session, seat)

    results = await asyncio.gather(reserve(1), reserve(1), return_exceptions=True)
    assert sum(1 for r in results if r is True) == 1

That test needs real committed transactions to produce a real conflict, so it uses a clean_database fixture that truncates afterwards rather than the rollback fixture. Keeping the two kinds of test visibly separate — most tests on the fast rollback fixture, a handful of concurrency tests on a genuinely committing one — is clearer than trying to make one fixture serve both.

Resolving Warnings, Errors & Common Mistakes

Advanced: Testing Code That Manages Its Own Transactions

Some code legitimately controls transaction boundaries — a job that commits per batch, a retry loop that rolls back and tries again, a saga step that must be durable before the next one starts. Those are exactly the behaviours worth testing, and they interact with the fixture in ways worth understanding.

A batch job that commits per chunk works unchanged: each commit releases a savepoint, a new one opens, and the outer rollback still discards everything. What the test cannot observe from inside the same transaction is durability — whether the rows would have survived a crash — because from the test's point of view they are equally invisible either way.

A retry loop that rolls back is more interesting, because a rollback inside a savepoint returns to the savepoint rather than discarding the whole test:

@pytest.mark.asyncio
async def test_retry_loop_recovers_after_conflict(session, monkeypatch):
    """The service's rollback returns to its savepoint; the test's data survives."""
    customer = Customer(email="ada@example.com")
    session.add(customer)
    await session.flush()

    attempts = {"n": 0}

    async def flaky_apply(session, customer_id: int) -> None:
        attempts["n"] += 1
        if attempts["n"] == 1:
            raise OperationalError("serialization failure", None, None)
        session.add(Order(customer_id=customer_id, total=decimal.Decimal("5")))

    await run_with_retries(session, flaky_apply, customer.id, retries=3)

    assert attempts["n"] == 2
    orders = (await session.execute(select(Order))).scalars().all()
    assert len(orders) == 1
    # The customer created before the failing attempt is still here: the rollback
    # returned to the service's savepoint, not to the start of the test.
    assert (await session.get(Customer, customer.id)) is not None

That last assertion is the one worth writing explicitly, because it encodes the behaviour the retry loop depends on: a failed attempt must not discard work done before it. A retry loop that rolls back the whole transaction instead of a savepoint will pass a naive test and lose data in production, and this is where that difference becomes visible.

For a true serialisation failure — the kind SERIALIZABLE produces and covered in setting transaction isolation level per session — the retry must re-read everything, which means it cannot run inside the failed transaction at all. Testing that path needs two real sessions and a committing fixture, as in the concurrency example above.

Keeping the Fixture Honest

A fixture that silently stops isolating tests is worse than no fixture, because the suite keeps passing. Two cheap checks keep it honest.

The first is a canary test that deliberately writes a row and asserts, in a separate test, that the row is gone. Ordering between tests is not guaranteed, so express it as one test that writes and one that counts, with the count asserting zero:

@pytest.mark.asyncio
async def test_canary_writes_a_row(session):
    session.add(Customer(email="canary@example.com"))
    await session.commit()


@pytest.mark.asyncio
async def test_canary_row_did_not_survive(session):
    found = await session.scalar(
        select(func.count()).select_from(Customer).where(
            Customer.email == "canary@example.com"
        )
    )
    assert found == 0, "a committed row survived the rollback fixture"

If someone later changes the fixture to bind the session to the engine instead of the connection — a one-word edit that looks harmless — this pair fails immediately and names the problem.

The second check is a lint rule, or simply a grep in CI, forbidding async_sessionmaker( and create_async_engine( outside conftest.py and the application's composition root. Every leak listed earlier reduces to a session created somewhere it should not have been, and a grep catches all of them before the suite has to.

Neither check is elaborate, and together they turn "the tests are isolated" from an assumption into something the suite actually asserts.

Frequently Asked Questions

Related

• Testing Async SQLAlchemy Applications — The parent guide: fixture layout, event-loop scoping, and the SQLite-versus-PostgreSQL split.
• Using pytest-asyncio fixtures with AsyncSession — Loop scoping and fixture ordering, and the errors each mismatch produces.
• Running tests against a PostgreSQL testcontainer — A real database per session, for the tests rollback isolation cannot serve alone.
• Setting transaction isolation level per session — Why a serialisation-failure retry cannot run inside the transaction that failed.