SQLAlchemy 2.0 Testing: Async Sessions, pytest-asyncio, and Transactional Fixtures

SQLAlchemy 2.0 Testing: Async Sessions, pytest-asyncio, and Transactional Fixtures

SQLAlchemy 2.0 shipped a fully unified async API that changes how you write database tests. If you're still using the 1.x synchronous patterns—or worse, relying on mock objects—you're leaving real bugs in production. This guide covers the correct way to test async SQLAlchemy code using pytest-asyncio, transactional fixtures, and real database connections.

Why Async SQLAlchemy Testing Is Different

The synchronous ORM has decades of testing patterns behind it. Async does not. SQLAlchemy 2.0's async layer (AsyncSession, AsyncEngine, AsyncConnection) requires a different fixture architecture because:

  • You cannot use synchronous pytest fixtures with coroutines without explicit handling
  • AsyncSession doesn't support rollback() the same way Session does in all contexts
  • The event loop must be shared between fixtures and tests—wrong setup causes ScopeMismatch errors

Getting this right from the start saves weeks of debugging flaky tests.

Setting Up the Async Engine

Start with a test-only async engine. Use an in-memory SQLite database for unit tests, and a real PostgreSQL instance (via Docker or Testcontainers) for integration tests.

# conftest.py
import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

# Use SQLite in-memory for speed; swap to PostgreSQL for integration tests
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"

@pytest_asyncio.fixture(scope="session")
async def async_engine():
    engine = create_async_engine(TEST_DATABASE_URL, echo=False)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()

The scope="session" here creates the schema once. Individual test functions get their own sessions scoped to function level—this is the key pattern for test isolation.

The Transactional Fixture Pattern

The most important pattern for fast SQLAlchemy tests is the nested transaction rollback. Each test runs inside a transaction that gets rolled back, leaving the database clean without dropping and recreating tables.

@pytest_asyncio.fixture
async def db_session(async_engine):
    async with async_engine.connect() as connection:
        await connection.begin()
        async_session = async_sessionmaker(
            bind=connection,
            expire_on_commit=False,
            class_=AsyncSession,
        )
        async with async_session() as session:
            yield session
        await connection.rollback()

This fixture:

  1. Opens a connection and begins a transaction
  2. Creates a session bound to that connection
  3. Yields the session to the test
  4. Rolls back the transaction after the test exits

The result: zero database state leaks between tests, even when tests commit() inside them—because the outermost transaction is rolled back.

Configuring pytest-asyncio

Install the required packages:

pip install pytest-asyncio aiosqlite sqlalchemy[asyncio]

Configure pytest.ini or pyproject.toml:

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

With asyncio_mode = "auto", every async test function is automatically treated as a coroutine test. You don't need @pytest.mark.asyncio on every function—a common source of forgotten decorators.

Writing Your First Async Test

Here's a complete example with a User model and a repository pattern:

# models.py
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column
from conftest import Base

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)
    name: Mapped[str] = mapped_column(String(100))
# test_user_repository.py
import pytest
from sqlalchemy import select
from models import User

async def test_create_user(db_session):
    user = User(email="test@example.com", name="Test User")
    db_session.add(user)
    await db_session.commit()

    result = await db_session.execute(select(User).where(User.email == "test@example.com"))
    fetched = result.scalar_one()

    assert fetched.name == "Test User"
    assert fetched.id is not None

async def test_users_isolated_between_tests(db_session):
    # This test runs in its own transaction — no user from the previous test is visible
    result = await db_session.execute(select(User))
    users = result.scalars().all()
    assert len(users) == 0

Both tests pass because the transactional fixture rolls back after each test. The second test genuinely sees an empty database.

Handling expire_on_commit=False

By default, SQLAlchemy expires all ORM attributes after a commit(). With async sessions, accessing an expired attribute triggers a lazy load—which fails with MissingGreenlet if you're outside an async context.

The fix: set expire_on_commit=False in your test session factory (shown above). For production sessions, handle this explicitly by eagerly loading what you need before committing.

async def test_user_attributes_after_commit(db_session):
    user = User(email="post@example.com", name="Post Commit")
    db_session.add(user)
    await db_session.commit()

    # With expire_on_commit=False, this works without re-querying
    assert user.email == "post@example.com"

Scoping: Session vs Function vs Module

Use the right scope for each fixture:

Fixture Scope Why
async_engine session Schema creation is expensive; do once
db_session function Each test needs isolated state
Seeded reference data module Static lookup tables shared across tests

Avoid session-scoped sessions unless your tests are explicitly read-only. State leaks in session-scoped sessions create the hardest-to-reproduce bugs.

Common Errors and Fixes

ScopeMismatch: async_generator function 'db_session' with scope 'function' can't be used in a fixture with scope 'session'

You have a session-scoped fixture trying to use a function-scoped one. Ensure parent fixtures have equal or wider scope.

greenlet_spawn has not been called

You're running sync code that tries to access a lazy-loaded ORM attribute. Fix: use selectinload() or joinedload() to eager-load relationships, or access attributes before exiting the async context.

Event loop is closed

Your event loop fixture scope doesn't match the engine fixture scope. When using asyncio_mode = "auto", pytest-asyncio manages the loop per test by default. If you need a session-scoped engine, you need a session-scoped event loop:

@pytest.fixture(scope="session")
def event_loop():
    import asyncio
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()

Testing with Real PostgreSQL

For integration tests, swap the engine URL and use Testcontainers:

@pytest_asyncio.fixture(scope="session")
async def async_engine():
    from testcontainers.postgres import PostgresContainer
    with PostgresContainer("postgres:16") as pg:
        url = pg.get_connection_url().replace("postgresql://", "postgresql+asyncpg://")
        engine = create_async_engine(url)
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
        yield engine
        await engine.dispose()

This gives you a real PostgreSQL instance per test session with zero manual setup. The container is destroyed when tests finish.

Integrating with HelpMeTest

Once your SQLAlchemy async tests are solid locally, add end-to-end coverage at the API layer with HelpMeTest. Write tests in plain English that exercise database-backed endpoints, and HelpMeTest will run them on a schedule against your deployed environment—catching regressions that unit tests miss.

Go To  https://api.example.com/users
POST   {"email": "health@example.com", "name": "Health Check"}
Status Code Should Be  201

The combination of fast transactional unit tests and scheduled integration monitoring gives you full coverage without the flakiness.

Summary

SQLAlchemy 2.0 async testing requires:

  1. A session-scoped AsyncEngine with schema creation
  2. Function-scoped sessions using nested transaction rollback for isolation
  3. asyncio_mode = "auto" in pytest config
  4. expire_on_commit=False to avoid MissingGreenlet errors
  5. Eager loading for relationships accessed after await session.commit()

These patterns make async database tests as fast and reliable as their synchronous counterparts.

Read more