Writing integration tests for asynchronous Python applications (like FastAPI using SQLAlchemy and asyncpg) is challenging. If multiple test cases write to the same database concurrently, they can cause test collisions, leading to flakey build pipelines. Standard solutions recreate the database schema for every test, but this slows down the CI process. A faster design runs each test case inside its own database transaction and rolls it back on completion.

By defining custom async pytest fixtures that handle transaction rollbacks, we can ensure database test isolation.

1. Implementing Async Transaction Rollback Fixtures

We write an async pytest fixture that creates a nested database transaction, yields the session, and rolls back all modifications:

# conftest.py
import pytest
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost:5432/test_db"
engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

@pytest.fixture
async def db_session():
    # Establish connection
    async with engine.connect() as conn:
        # Begin transaction
        async with conn.begin() as transaction:
            async_session = AsyncSessionLocal(bind=conn)
            
            yield async_session
            
            # Roll back all database writes
            await transaction.rollback()
            await async_session.close()

2. Writing Clean Integration Tests

Using this fixture, developers write standard async tests. Since all inserts are rolled back, tests leave the database clean, preventing test collisions.

Savepoint-Wrapped AsyncPG Test Fixtures with Pytest-Asyncio

Bind each async test function to an isolated PostgreSQL transaction savepoint that rolls back automatically on teardown, enabling parallel test execution with pytest-xdist.