Python Testing & Debugging Beyond unittestAdvanced Python testing playbooks
Integration & Data

Rolling Back Every Test with Nested Transactions

Application code commits. It calls session.commit() at the end of a unit of work because that is what application code is supposed to do, and a naive rollback fixture is defeated by the first one: the commit ends the fixture's transaction, the rows become durable, and the next test inherits them. The fix is to give the session a savepoint rather than the outer transaction, so the application's commit releases the savepoint and the fixture's rollback still undoes everything.

Prerequisites

  • SQLAlchemy >= 2.0, which introduced join_transaction_mode. On 1.4 the same effect needs a manual after_transaction_end listener.
  • A database engine with transactional semantics for the statements under test; Postgres qualifies, including DDL.
  • pytest >= 8.0, and the layering described in database fixtures and transactional tests.

Solution

Python
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session


@pytest.fixture(scope="session")
def engine(postgres_dsn):
    engine = create_engine(postgres_dsn, pool_pre_ping=True)
    yield engine
    engine.dispose()


@pytest.fixture
def db_session(engine):
    # One connection, one outer transaction — the thing we will roll back.
    connection = engine.connect()
    outer = connection.begin()

    # Bound to THAT connection, not the engine, and joined via savepoints:
    # session.commit() now issues RELEASE SAVEPOINT rather than COMMIT.
    session = Session(bind=connection, join_transaction_mode="create_savepoint")
    try:
        yield session
    finally:
        session.close()
        outer.rollback()           # undoes every "commit" the test made
        connection.close()
Python
def test_order_is_persisted(db_session):
    create_order(db_session, customer_id="cus_1", total=1234)   # commits inside
    assert db_session.query(Order).count() == 1


def test_database_is_clean(db_session):
    # Passes in either order only if the previous test's commit was contained.
    assert db_session.query(Order).count() == 0
Engine-bound versus connection-bound sessions Two arrangements. A session bound to the engine checks out its own connection from the pool, so its commit is a real commit outside the fixture's transaction and the rows survive. A session bound to the fixture's connection with savepoints runs inside the outer transaction, so its commit releases a savepoint and the fixture's rollback removes everything. The binding decides whether the rollback means anything Session(bind=engine) fixture: connection A, BEGIN session: checks out connection B commit on B → really committed rollback on A → nothing to undo rows leak into the next test Session(bind=connection, savepoint) fixture: connection A, BEGIN session: SAVEPOINT on A commit → RELEASE SAVEPOINT rollback on A → undoes it all every test starts clean
The left-hand arrangement looks correct and is the one most often found in older conftest.py files. Its failure appears as order dependence, far from the fixture.

Why this works

A savepoint is a named marker inside an open transaction. RELEASE SAVEPOINT discards the marker and keeps the work — but only as far as the enclosing transaction's eventual fate. With join_transaction_mode="create_savepoint", SQLAlchemy opens a savepoint whenever the session begins its own transaction on a connection that already has one, and translates commit() into a release. The application sees normal commit semantics: later queries in the same test see the committed rows. The database sees a transaction that was never committed and is finally rolled back.

Binding to the connection is the half that is easy to get wrong. A session bound to the engine asks the pool for a connection, gets a different one, and commits there — entirely outside the fixture's transaction.

Edge cases and failure modes

  • Code that creates its own Session(engine). It escapes the fixture. Inject the session, or point the application's session factory at the fixture's connection in tests.
  • A second connection must see the data. A live server or a background thread uses its own connection, and uncommitted data is invisible across connections by definition. Those tests need real commits and a truncation fixture.
  • DDL on MySQL. CREATE TABLE commits implicitly and ends the outer transaction. Postgres does not have this problem.
  • session.rollback() inside the application. It rolls back to the savepoint, which is correct, and the session opens a new savepoint for subsequent work.
  • Legacy 1.4 recipes. The after_transaction_end listener that restarted a nested transaction is no longer needed on 2.0 and can conflict with create_savepoint. Delete it.

What the SQL actually looks like

Turning on statement logging for one test makes the mechanism concrete, and it is the fastest way to confirm a fixture is doing what it claims.

Python
import logging

logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)
Plain text
BEGIN (implicit)                                  -- fixture: outer transaction
SAVEPOINT sa_savepoint_1                          -- session joins via savepoint
INSERT INTO "order" (customer_id, total) VALUES ('cus_1', 1234)
RELEASE SAVEPOINT sa_savepoint_1                  -- application's commit()
SAVEPOINT sa_savepoint_2                          -- next unit of work
SELECT count(*) FROM "order"                      -- sees the committed row
RELEASE SAVEPOINT sa_savepoint_2
ROLLBACK                                          -- fixture teardown: all gone

Every COMMIT the application thought it issued is a RELEASE SAVEPOINT, and the only transaction-ending statement is the final ROLLBACK. If the log shows a bare COMMIT anywhere, something opened a connection outside the fixture, and the log line immediately above it usually names the query that did it.

Statements issued across one test A vertical sequence of SQL statements inside one outer transaction. Two savepoints open and are released as the application commits twice, while queries in between see the committed rows. The final statement is a rollback of the outer transaction, which removes everything the test wrote. Two "commits", zero COMMIT statements BEGIN — outer transaction, owned by the fixture SAVEPOINT sa_1 INSERT INTO "order" … RELEASE SAVEPOINT sa_1 ← application commit() SAVEPOINT sa_2 SELECT count(*) → 1 RELEASE SAVEPOINT sa_2 ← the row is visible here ROLLBACK — every row above disappears
A bare COMMIT anywhere in this log is the leak. Grepping the test run's SQL log for it is a thirty-second audit of the whole suite.

That grep generalises into a permanent guard. A small event listener that fails the test if a real COMMIT reaches the database — outside the handful of tests marked as needing committed data — catches every future regression of the binding at the moment it is introduced, rather than as a mysterious order dependence weeks later:

Python
from sqlalchemy import event


@pytest.fixture(autouse=True)
def forbid_real_commits(request, engine):
    if request.node.get_closest_marker("committed_data"):
        yield
        return

    def on_commit(conn):
        raise AssertionError("a real COMMIT escaped the transactional fixture")

    event.listen(engine, "commit", on_commit)
    yield
    event.remove(engine, "commit", on_commit)

The listener fires on the engine's commit event, which a savepoint release does not trigger, so it is silent for correctly routed sessions and loud for anything that bypassed them. Pairing it with an explicit committed_data marker for the tests that genuinely need durable rows keeps the exception visible in the code rather than scattered through fixtures.

The async variant

The async engine follows the same structure with awaits in the obvious places, plus one extra constraint: the connection belongs to a loop.

Python
import pytest_asyncio
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine


@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def async_engine(postgres_dsn):
    engine = create_async_engine(postgres_dsn.replace("postgresql://", "postgresql+asyncpg://"))
    yield engine
    await engine.dispose()


@pytest_asyncio.fixture(loop_scope="session")
async def async_session(async_engine):
    async with async_engine.connect() as connection:
        outer = await connection.begin()
        session = AsyncSession(bind=connection, join_transaction_mode="create_savepoint")
        try:
            yield session
        finally:
            await session.close()
            await outer.rollback()

loop_scope="session" on both fixtures is not decoration. The engine's pool holds connections registered with the session loop; a per-test fixture running on a function loop would try to use them from a different loop and fail with the errors catalogued in sharing an event loop across a test module.

Pointing the application at the fixture's session

Most real applications do not receive a session as a function argument; they get one from a factory, a dependency-injection container or a framework hook. The fixture has to reach that mechanism, or the application silently writes outside the transaction.

Python
import pytest

from myapp import db


@pytest.fixture(autouse=True)
def route_app_sessions(db_session, monkeypatch):
    # The application calls db.get_session(); in tests it gets the fixture's.
    monkeypatch.setattr(db, "get_session", lambda: db_session)
Python
# FastAPI: override the dependency rather than patching a module.
import pytest
from fastapi.testclient import TestClient

from myapp.main import app, get_db


@pytest.fixture
def client(db_session):
    app.dependency_overrides[get_db] = lambda: db_session
    try:
        yield TestClient(app)
    finally:
        app.dependency_overrides.clear()

Both routes converge on the same property: every query the application makes goes through the one connection holding the outer transaction. The FastAPI version is preferable where it applies, because dependency overrides are a supported extension point rather than a patch — the argument made generally in dependency injection for testability.

Routing every application query through one connection The fixture opens one connection and an outer transaction. A dependency override or a patched session factory hands that session to the application's request handler and service layer, so all their queries share the connection and are undone by the single rollback at teardown. One connection, every query, one rollback db_session fixture connection A, BEGIN dependency override get_db → db_session request handler service layer Anything that opens its own Session(engine) falls outside this picture and leaks.
The pair of order-independent tests above is the check that this routing is complete. If either fails, something in the application found another way to the pool.

Cost, measured

The reason to go to this trouble is speed, and it is worth knowing the size of the effect. On a typical Postgres container, a test that truncates six tables in teardown spends thirty to sixty milliseconds on cleanup alone; the same test with this fixture spends well under one millisecond on its ROLLBACK. Across four hundred database tests that is the difference between roughly twenty seconds of pure cleanup and effectively none, before counting the setup savings from never recreating anything.

The less visible benefit is reliability. A truncation fixture that raises halfway — a foreign key it did not know about, a table added last week — leaves the database dirty and fails every subsequent test with an error unrelated to its cause. A rollback cannot half-fail in that way: either the transaction is rolled back, or the connection is dead and the next test gets a fresh one from the pool. That asymmetry is why the pattern survives in large suites long after the performance argument has been forgotten. It also means the fixture needs no maintenance when the schema grows: a new table is covered automatically, where a cleanup routine would need editing.

Frequently Asked Questions

Why do rows survive between tests even though the fixture rolls back? Usually because the session is bound to the engine rather than to the connection that owns the outer transaction, so it checks out a different connection from the pool and commits there. Bind the session to the specific connection on which the outer transaction was begun.

Does this work with the async SQLAlchemy engine? Yes. AsyncConnection.begin and AsyncSession with join_transaction_mode="create_savepoint" behave identically; the fixture becomes an async generator and the rollback is awaited. Match the fixture's loop scope to the engine's so the connection and its loop share a lifetime.

What about code that opens its own session? It bypasses the fixture entirely and writes outside the rolled-back transaction. Either inject the session so the code uses the fixture's, or configure the application's session factory in tests to bind to the fixture's connection.

← Back to Database Fixtures & Transactional Tests