Testing Async SQLAlchemy Applications

Wrap each test in an outer transaction that is rolled back at the end, bind the session to that connection with a nested SAVEPOINT, and every test starts from an identical database in single-digit milliseconds without truncating a single table. Getting there under asyncio means solving three problems the synchronous recipe never had — event-loop scoping, an engine that must outlive nothing, and expire_on_commit — and this page works through all three. It sits under mastering SQLAlchemy 2.0 Core and ORM architecture.

Concept & Execution Model

A database test needs three properties, and they pull against each other: isolation, so tests cannot see each other's writes; speed, so the suite runs often enough to be useful; and fidelity, so a passing test means the code works against the real database.

The rollback-per-test pattern gets all three. Each test runs inside a transaction that is never committed, so isolation is provided by the database's own MVCC rather than by cleanup code. Teardown is a single ROLLBACK, so speed is bounded by the test itself rather than by table truncation. And because the code under test runs against a real connection issuing real SQL, fidelity is whatever the database is.

The subtlety is that application code commits. A test whose subject calls session.commit() would, naively, commit the outer transaction and destroy the isolation. The resolution is the nested transaction: the session runs inside a SAVEPOINT, its commit releases that savepoint rather than the outer transaction, and a listener immediately opens a fresh one so subsequent work still has somewhere to go.

# conftest.py — the core of an async test suite
from __future__ import annotations

import asyncio
from collections.abc import AsyncIterator

import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import (
    AsyncConnection, AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine,
)

from myapp.models import Base

TEST_URL = "postgresql+asyncpg://app:secret@localhost:5432/orders_test"


@pytest_asyncio.fixture(scope="session")
async def engine() -> AsyncIterator[AsyncEngine]:
    # Built inside a fixture, never at import time: an engine created at module
    # level exists before pytest-asyncio has installed an event loop.
    engine = create_async_engine(TEST_URL, poolclass=None, echo=False)
    async with engine.begin() as connection:
        await connection.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()


@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()          # discards everything the test did


@pytest_asyncio.fixture
async def session(connection: AsyncConnection) -> AsyncIterator[AsyncSession]:
    factory = async_sessionmaker(
        bind=connection,
        expire_on_commit=False,               # assertions after commit must not refresh
        join_transaction_mode="create_savepoint",
    )
    async with factory() as session:
        yield session

join_transaction_mode="create_savepoint" is what makes application-level commits safe: SQLAlchemy 2.0 handles the savepoint bookkeeping that older recipes did with a manual after_transaction_end event listener. If you are maintaining one of those older conftest.py files, this argument replaces the listener entirely.

Test Construction & Async Execution Patterns

With the fixture in place, tests are ordinary async functions. The session they receive behaves exactly like a production one — including committing.

# Async — a test whose subject commits, with no special handling
import decimal

import pytest
from sqlalchemy import select

from myapp.models import Customer, Order
from myapp.services import place_order


@pytest.mark.asyncio
async def test_place_order_persists_and_totals(session):
    customer = Customer(email="ada@example.com")
    session.add(customer)
    await session.flush()

    # The service under test commits. That releases a savepoint, not the outer
    # transaction, so the test stays isolated.
    order = await place_order(session, customer.id, decimal.Decimal("42.00"))

    stored = (
        await session.execute(select(Order).where(Order.id == order.id))
    ).scalar_one()
    assert stored.total == decimal.Decimal("42.00")
    assert stored.customer_id == customer.id

Two conventions repay themselves quickly. Build test data through the same constructors the application uses rather than through raw INSERT statements, so a model change breaks the test loudly instead of leaving it asserting against a schema that no longer exists. And assert by re-querying rather than by inspecting the object you just created — the object in memory can be correct while the row is not, which is precisely the class of bug an integration test exists to catch.

# Sync — the same shape, for a codebase still running a synchronous suite
def test_place_order_sync(sync_session):
    customer = Customer(email="ada@example.com")
    sync_session.add(customer)
    sync_session.flush()
    order = place_order_sync(sync_session, customer.id, decimal.Decimal("42.00"))
    stored = sync_session.execute(
        select(Order).where(Order.id == order.id)
    ).scalar_one()
    assert stored.total == decimal.Decimal("42.00")

Testing a FastAPI application adds one step: the dependency that normally builds a session per request must be overridden to hand back the test's session, so the request runs inside the same transaction the fixture will roll back. The dependency shape itself is covered in integrating SQLAlchemy async with FastAPI and Starlette.

@pytest_asyncio.fixture
async def client(session) -> AsyncIterator[AsyncClient]:
    from httpx import ASGITransport, AsyncClient
    from myapp.main import app, get_session

    app.dependency_overrides[get_session] = lambda: session
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://t") as c:
        yield c
    app.dependency_overrides.clear()

State Management & Session Boundaries in Tests

The most common way an async test suite goes wrong is a mismatch between the lifetime of a connection and the lifetime of an event loop. asyncio connections belong to the loop that created them; hand one to a different loop and you get RuntimeError: Task got Future attached to a different loop, from a frame that explains nothing.

The rule is simple to state: anything holding a connection must not outlive the event loop it was created on. In practice that means the event loop fixture and the engine fixture must have the same scope. With pytest-asyncio in modern configuration, that is one line of configuration rather than a custom fixture:

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

With a session-scoped loop, a session-scoped engine is safe. With the default function-scoped loop, a session-scoped engine is not — and the failure is intermittent, appearing only once a second test reuses the pooled connection.

The second boundary question is expire_on_commit. Left at its default, every attribute of every instance is expired when the test's subject commits, so the assertion that follows triggers a refresh — which, on an AsyncSession, raises rather than reloading. Setting expire_on_commit=False on the test factory removes the problem and matches what production code should be doing anyway, for the reasons set out in using expire_on_commit=False in FastAPI dependencies.

A third, subtler boundary: the fixture's session and the code under test must be the same session. A service that builds its own session from a global factory will get a connection from the pool rather than the test's connection, and its writes will land in a different transaction — one that really does commit, leaving residue and breaking the next test. Services should take a session as an argument; that they must, in order to be testable, is a design benefit rather than a cost.

Advanced: Parallel Runs, Factories, and Schema Setup

Once a suite is fast, the next constraint is usually wall-clock time across the whole run, and the answer is parallelism. Rollback isolation makes tests independent, which makes pytest -n auto safe — with one caveat: parallel workers must not share a database, because DDL and sequences are not transactional.

# conftest.py — one database per xdist worker
import os


@pytest_asyncio.fixture(scope="session")
async def engine() -> AsyncIterator[AsyncEngine]:
    worker = os.environ.get("PYTEST_XDIST_WORKER", "gw0")
    url = f"postgresql+asyncpg://app:secret@localhost:5432/orders_test_{worker}"
    engine = create_async_engine(url)
    async with engine.begin() as connection:
        await connection.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()

Creating the schema with Base.metadata.create_all is fast and is the right choice for a unit suite. It is the wrong choice for a suite meant to catch migration problems, because it builds the schema the models describe rather than the schema the migrations produce — and those diverge exactly when a migration is wrong. Run the migrations instead for at least one job in CI:

async def _migrate(engine: AsyncEngine) -> None:
    from alembic import command
    from alembic.config import Config

    config = Config("alembic.ini")
    config.set_main_option("sqlalchemy.url", str(engine.url.render_as_string(False)))
    await asyncio.to_thread(command.upgrade, config, "head")

Test data is the other place a suite quietly rots. Constructing entities inline is fine for two fields and unmanageable for twenty, and the usual answer is a factory function per aggregate with sensible defaults and explicit overrides:

def make_order(**overrides) -> Order:
    defaults = dict(
        reference=f"ORD-{uuid.uuid4().hex[:8]}",
        total=decimal.Decimal("10.00"),
        status="pending",
    )
    return Order(**(defaults | overrides))

The discipline that keeps factories useful: a test should override exactly the fields it is about, and nothing else. A test that overrides five fields when it cares about one is hiding its own subject, and it will keep passing after the behaviour it was written to protect has been broken.

Hybrid Strategies: SQLite for Speed, PostgreSQL for Truth

Most projects end up with two suites rather than one, and the split is worth making deliberately rather than by accident.

An in-memory SQLite suite runs in milliseconds and needs no infrastructure, which makes it the right home for anything that is really about Python: service logic, validation, mapping, query construction. Its configuration has one non-obvious requirement — StaticPool, without which every connection gets a fresh, empty database — covered in using aiosqlite for async tests and local development.

A PostgreSQL suite in a container is the right home for anything that is really about the database: migrations, constraints, isolation levels, concurrency, native types, and anything using PostgreSQL-specific SQL. That list is longer than it first appears — a query using ON CONFLICT, a JSONB column, a partial index, or a SERIALIZABLE retry loop cannot be meaningfully tested anywhere else.

# Two engines, chosen by marker — fast by default, faithful when it matters
import pytest


@pytest_asyncio.fixture(scope="session")
async def engine(request) -> AsyncIterator[AsyncEngine]:
    if request.node.get_closest_marker("postgres"):
        url = os.environ["TEST_DATABASE_URL"]
        engine = create_async_engine(url)
    else:
        from sqlalchemy.pool import StaticPool

        engine = create_async_engine(
            "sqlite+aiosqlite:///:memory:",
            poolclass=StaticPool,
            connect_args={"check_same_thread": False},
        )
    async with engine.begin() as connection:
        await connection.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()

The failure mode to guard against is a suite that runs entirely on SQLite and is believed to prove more than it does. SQLite serialises writers, so no deadlock or serialisation-failure path is ever exercised; it degrades JSONB and ARRAY to text, so a schema PostgreSQL would reject can pass; and it has no deferrable constraints, so a test relying on deferred checking proves nothing. A @pytest.mark.postgres marker on the tests that need the real thing, plus a CI job that runs them, is the whole discipline.

Production Pitfalls & Anti-Patterns

• RuntimeError: Task <...> got Future <...> attached to a different loop — a connection created on one event loop is being used on another, almost always a session-scoped engine with a function-scoped loop. Give the loop and the engine the same scope.
• sqlalchemy.exc.InterfaceError: cannot perform operation: another operation is in progress — two coroutines sharing one AsyncSession. A test that gathers concurrent work must give each task its own session, exactly as production must.
• MissingGreenlet in an assertion — the assertion touched a relationship nothing loaded. This is a finding about the code under test, not about the fixture: the production path would fail the same way. Add the loader option to the query, not a workaround to the test.
• Tests pass alone and fail in a suite — something is escaping the rollback. The usual cause is a service that built its own session from a global factory instead of accepting one, so its writes went to a different connection and really did commit.
• sqlite3.OperationalError: no such table after a fixture that clearly created it — an in-memory SQLite engine without StaticPool, so each connection got its own empty database.
• A suite that truncates tables between tests — works, and costs tens of milliseconds per test plus an ordering constraint the rollback approach does not have. Migrate to rollback isolation before optimising anything else.
• Schema built with create_all in the only CI job — the suite then tests the schema the models describe rather than the one the migrations produce, which is exactly the divergence a broken migration creates. Run at least one job against migrated schema.

Migrating an Existing Suite to Rollback Isolation

Most suites arrive at rollback isolation from somewhere worse — usually a TRUNCATE loop, sometimes a per-test database drop and recreate. The migration is worth doing incrementally, because the two approaches can coexist while it happens.

Start by adding the connection and session fixtures alongside whatever exists. Convert one test module to request session instead of the old fixture, and confirm it passes both alone and inside the full suite — the second run is the one that matters, because it is where escaped writes show up. Then convert module by module, deleting the truncation fixture only when nothing requests it.

Two problems reliably surface during that conversion, and both are worth fixing rather than working around. The first is code that builds its own session; those call sites have to start accepting one, which is a small refactor with a large payoff. The second is tests that depended on data committed by an earlier test — an implicit ordering that truncation-based suites tolerate and rollback isolation exposes immediately. Those tests were already fragile; making each one create the data it needs is the fix, and a factory function keeps that from being verbose.

Expect the suite to get faster by roughly the truncation cost per test, which for a schema of any size is tens of milliseconds. On a thousand-test suite that is a minute of wall clock recovered on every run, which changes how often people run it — and a suite that gets run is worth considerably more than one that is merely correct.

Frequently Asked Questions

Related

• Mastering SQLAlchemy 2.0 Core and ORM Architecture — The parent guide covering the session and execution model these fixtures are built on.
• Rolling back database state between async tests — The nested-transaction fixture in full, including the savepoint listener and what replaced it.
• Using pytest-asyncio fixtures with AsyncSession — Event-loop scoping, fixture ordering, and the errors each mismatch produces.
• Running tests against a PostgreSQL testcontainer — Starting a real database per session, and what it lets you test that SQLite cannot.
• Using aiosqlite for async tests and local development — The StaticPool requirement and the four things SQLite can never prove.