SQLAlchemy 2.x Testing

Best practices for testing SQLAlchemy ORM code (sync and async) with pytest.

Agent Quick Path

Use this path first; read deeper sections only when needed.

Create engine once per test session.
Open connection + outer transaction per test function.
Bind Session/AsyncSession to that connection with join_transaction_mode="create_savepoint".
Let code under test call commit() safely; rollback outer transaction after test.
Inject session into FastAPI via dependency override and always clear overrides.

Branching logic:

Sync stack: use create_engine + Session fixtures.
Async stack: use create_async_engine + AsyncSession fixtures + pytest.mark.anyio.
SQLite in-memory with threaded client: use StaticPool when required by framework threading behavior.
Async relationship access fails (MissingGreenlet): eager load (selectinload) or explicit refresh.

Core Concepts

Concept	Preferred approach
Isolate tests from production DB	Use an in-memory SQLite engine per test
Prevent data leaking between tests	Roll back at the connection level after each test
Inject test DB into FastAPI	Override the `get_db` / `get_session` dependency
Create schema for tests	Call `Base.metadata.create_all(engine)` once per session
Avoid lazy-load errors in async	Use `expire_on_commit=False`; use `selectinload()` for relationships

Sync SQLAlchemy + FastAPI (Recommended Pattern)

The canonical 2.0 pattern joins a test Session into an external transaction on a shared Connection, then rolls back after each test. This means session.commit() calls within the code under test are "committed" to a savepoint, not the real transaction, and are fully undone after the test.

# tests/conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from fastapi.testclient import TestClient

from myapp.main import app
from myapp.db.session import get_db
from myapp.db.models import Base

@pytest.fixture(scope="session")
def engine():
    e = create_engine("sqlite://", connect_args={"check_same_thread": False})
    Base.metadata.create_all(bind=e)
    yield e
    Base.metadata.drop_all(bind=e)

@pytest.fixture()
def db_session(engine):
    connection = engine.connect()
    transaction = connection.begin()
    session = Session(bind=connection, join_transaction_mode="create_savepoint")
    yield session
    session.close()
    transaction.rollback()
    connection.close()

@pytest.fixture()
def client(db_session):
    app.dependency_overrides[get_db] = lambda: db_session
    yield TestClient(app)
    app.dependency_overrides = {}

Key points:

join_transaction_mode="create_savepoint" means each session.commit() inside code under test issues a SAVEPOINT release, not a real COMMIT — everything is rolled back when the test ends.
scope="session" engine + scope="function" session/connection gives fast table creation with full per-test isolation.
SQLite in-memory (sqlite://) is preferred: no files, no cleanup, fast.

Async SQLAlchemy + FastAPI

For AsyncSession / AsyncEngine, the setup mirrors the sync version but uses async fixtures and pytest-anyio.

# tests/conftest.py
import pytest
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from fastapi.testclient import TestClient  # sync client still works
from httpx import AsyncClient, ASGITransport

from myapp.main import app
from myapp.db.session import get_db
from myapp.db.models import Base

@pytest.fixture(scope="session")
def anyio_backend():
    return "asyncio"

@pytest.fixture(scope="session")
async def async_engine():
    e = create_async_engine("sqlite+aiosqlite://", connect_args={"check_same_thread": False})
    async with e.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield e
    async with e.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await e.dispose()

@pytest.fixture()
async def async_session(async_engine):
    async with async_engine.connect() as conn:
        await conn.begin()
        session = AsyncSession(bind=conn, join_transaction_mode="create_savepoint",
                               expire_on_commit=False)
        yield session
        await session.close()
        await conn.rollback()

@pytest.fixture()
async def async_client(async_session):
    async def override_get_db():
        yield async_session

    app.dependency_overrides[get_db] = override_get_db
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
        yield ac
    app.dependency_overrides = {}

Notes:

Async fixtures need @pytest.mark.anyio on the test or anyio_backend session fixture.
expire_on_commit=False prevents expired-attribute access on objects after await session.commit().
Install: aiosqlite, anyio[asyncio], httpx.

SQLModel Pattern (FastAPI + SQLModel)

SQLModel's official test pattern uses StaticPool + in-memory SQLite, with separate session and client pytest fixtures.

import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool

from myapp.main import app, get_session  # get_session is the SQLModel dependency

@pytest.fixture(name="session")
def session_fixture():
    engine = create_engine(
        "sqlite://",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session

@pytest.fixture(name="client")
def client_fixture(session: Session):
    app.dependency_overrides[get_session] = lambda: session
    yield TestClient(app)
    app.dependency_overrides.clear()

# --- Tests ---

def test_create_hero(client: TestClient):
    response = client.post("/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"})
    assert response.status_code == 200

def test_read_heroes(session: Session, client: TestClient):
    # Directly insert test data via the session — no HTTP call needed for setup
    from myapp.models import Hero
    session.add(Hero(name="Test Hero", secret_name="Hidden"))
    session.commit()
    response = client.get("/heroes/")
    assert response.status_code == 200
    assert len(response.json()) == 1

Rules:

Use StaticPool so a single in-memory SQLite connection is shared across threads (required by TestClient's threading model).
Both the client fixture and test functions can receive the same session — insert data directly for controlled setup rather than via the API.
Always call app.dependency_overrides.clear() in fixture teardown (after yield).

Async Session: Avoiding Implicit I/O

SQLAlchemy async has strict rules about lazy loading — attributes that would trigger IO on access will raise an error.

# WRONG — will raise MissingGreenlet / lazy load error
result = await session.execute(select(User))
user = result.scalars().one()
print(user.posts)  # lazy load, fails in async context

# RIGHT — use selectinload() to load relationships eagerly
from sqlalchemy.orm import selectinload

result = await session.execute(
    select(User).options(selectinload(User.posts))
)
user = result.scalars().one()
print(user.posts)  # already loaded, no IO needed

Other strategies:

AsyncAttrs mixin: access any attribute as an awaitable via await obj.awaitable_attrs.relationship_name.
write_only relationships: never loaded implicitly; queried explicitly.
await session.refresh(obj, ["attribute_name"]): force-load a specific attribute after the fact.

Fixture Scope Decision Table

What to scope	`scope`	Reason
Engine + DDL (`create_all`)	`session`	Expensive; shared across all tests
Connection + Transaction	`function`	Rolled back per test for isolation
Session	`function`	One transaction per test
TestClient / AsyncClient	`function`	Depends on session; recreated per test

Common Pitfalls

Problem	Cause	Fix
`MissingGreenlet` in async	Lazy-loaded relationship accessed outside awaitable context	Use `selectinload()` or `AsyncAttrs.awaitable_attrs`
`RuntimeError: Event loop is closed`	`AsyncEngine` not disposed	Call `await engine.dispose()` in fixture teardown
Tests share state / data bleeds	Session not rolled back	Use `join_transaction_mode="create_savepoint"` + rollback pattern
`StaticPool` not used with SQLite in-memory	TestClient spawns threads that get separate in-memory DBs	Always add `poolclass=StaticPool` for in-memory SQLite
`expire_on_commit=True` (default) breaks async	Accessing attributes after commit triggers lazy IO	Set `expire_on_commit=False` on AsyncSession
Not resetting `dependency_overrides`	Override persists into next test	Always clear in fixture teardown, after `yield`

9.7 KiB Raw Blame History