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
pytestfixtures with coroutines without explicit handling AsyncSessiondoesn't supportrollback()the same waySessiondoes in all contexts- The event loop must be shared between fixtures and tests—wrong setup causes
ScopeMismatcherrors
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:
- Opens a connection and begins a transaction
- Creates a session bound to that connection
- Yields the session to the test
- 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) == 0Both 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 201The combination of fast transactional unit tests and scheduled integration monitoring gives you full coverage without the flakiness.
Summary
SQLAlchemy 2.0 async testing requires:
- A session-scoped
AsyncEnginewith schema creation - Function-scoped sessions using nested transaction rollback for isolation
asyncio_mode = "auto"in pytest configexpire_on_commit=Falseto avoid MissingGreenlet errors- Eager loading for relationships accessed after
await session.commit()
These patterns make async database tests as fast and reliable as their synchronous counterparts.