# SQLAlchemy testing best practices

> How to test SQLAlchemy 2.0 code with pytest: schema once per session, savepoint rollback per test, factory_boy factories, statement counting, AsyncSession and Alembic checks.

SQLAlchemy doesn't ship a test framework, so every project builds its own from fixtures. The ones that work well make the same choices: a real database of the same kind as production, the schema created once, every test inside a transaction that is rolled back, factories for data, and a few tests that keep the queries and the migrations honest.

## Test against the database you run in production

If production runs PostgreSQL, run the tests on PostgreSQL. SQLite is tempting because it needs no server, but it differs in ways your tests will hide rather than catch:

- Types: no real `JSONB`, arrays, `UUID`, timezone-aware timestamps or enforced `VARCHAR` lengths.
- Constraints: foreign keys are off unless you enable `PRAGMA foreign_keys`, and many `ALTER TABLE` operations aren't supported.
- SQL: `ON CONFLICT`, `RETURNING`, window functions, `ILIKE`, locking (`SELECT ... FOR UPDATE`) and case sensitivity behave differently or not at all.
- Concurrency: SQLite locks the whole database, so race conditions and deadlocks never show up.

A local PostgreSQL in Docker and a `services:` container in CI cost little. [Make the test database server skip durability work](https://flakytest.dev/guides/fast-test-database-server/) shows how to make it nearly as fast as SQLite.

## Create the schema once, roll back every test

Creating tables is expensive: `metadata.create_all()` and `drop_all()` in a function-scoped fixture can cost more than the tests themselves. Create the schema once per session, then give each test a connection with an outer transaction and bind the `Session` with `join_transaction_mode="create_savepoint"`:

```python
# tests/conftest.py
import os

import pytest
from sqlalchemy import create_engine

from myapp.models import Base
from tests.factories import TestSession

DATABASE_URL = os.environ.get("TEST_DATABASE_URL", "postgresql+psycopg://localhost/myapp_test")


@pytest.fixture(scope="session")
def engine():
    engine = create_engine(DATABASE_URL)
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)
    yield engine
    engine.dispose()


@pytest.fixture
def session(engine):
    with engine.connect() as connection:
        transaction = connection.begin()
        TestSession.configure(bind=connection, join_transaction_mode="create_savepoint")
        session = TestSession()
        yield session
        TestSession.remove()
        transaction.rollback()
```

When the code under test calls `session.commit()`, it releases a savepoint instead of committing, and the outer `transaction.rollback()` undoes everything at the end of the test. The full recipe, with the reasons behind each line, is in [create the SQLAlchemy schema once, roll back per test](https://flakytest.dev/guides/sqlalchemy-rollback-per-test/).

Two habits keep this setup trustworthy:

- Everything the test touches must go through that one connection. A second engine, a background thread or a separate `sessionmaker` in the app opens its own connection, can't see the uncommitted rows and leaves its own rows behind.
- Don't `drop_all()` at the end of the session. Leaving the tables makes a failed run easy to inspect, and the next run starts by dropping them anyway.

## Use factories bound to the test session

factory_boy's `SQLAlchemyModelFactory` needs to know which session to add objects to. The factory_boy docs recommend a `scoped_session` shared by the factories and the tests, which is what `TestSession` above is:

```python
# tests/factories.py
import factory
from sqlalchemy.orm import scoped_session, sessionmaker

from myapp.models import Order, User

TestSession = scoped_session(sessionmaker())


class UserFactory(factory.alchemy.SQLAlchemyModelFactory):
    class Meta:
        model = User
        sqlalchemy_session = TestSession
        sqlalchemy_session_persistence = "flush"

    email = factory.Sequence(lambda n: f"user{n}@example.com")


class OrderFactory(factory.alchemy.SQLAlchemyModelFactory):
    class Meta:
        model = Order
        sqlalchemy_session = TestSession
        sqlalchemy_session_persistence = "flush"

    customer = factory.SubFactory(UserFactory)
```

Pass the `scoped_session` itself, not `TestSession()`. `sqlalchemy_session_persistence = "flush"` sends the `INSERT` so the rows have primary keys, without committing. Avoid `"commit"`: with the savepoint setup it still works, but it hides the fact that the test never needed a commit. factory_boy 3.3 also accepts `sqlalchemy_session_factory`, a callable returning the session, if you prefer that to a scoped session.

Use `UserFactory.build()` for objects that never need to reach the database, and `create_batch()` only when the test needs many rows. [Faster test factories](https://flakytest.dev/guides/faster-test-factories/) covers the rest.

## Count statements and catch N+1 queries

ORM code that is correct can still issue a query per row. Lock the number of statements in with a test. SQLAlchemy's `before_cursor_execute` event sees every statement sent to the database:

```python
# tests/utils.py
from contextlib import contextmanager

from sqlalchemy import event


@contextmanager
def count_statements(engine):
    statements = []

    def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
        statements.append(statement)

    event.listen(engine, "before_cursor_execute", before_cursor_execute)
    try:
        yield statements
    finally:
        event.remove(engine, "before_cursor_execute", before_cursor_execute)
```

```python
from myapp.orders import list_orders
from tests.factories import OrderFactory
from tests.utils import count_statements


def test_list_orders_loads_lines_eagerly(session, engine):
    OrderFactory.create_batch(10)
    session.expire_all()

    with count_statements(engine) as statements:
        orders = list_orders(session)
        [line.product_id for order in orders for line in order.lines]

    assert len(statements) == 2  # orders, then lines with selectinload()
```

`session.expire_all()` matters: without it the objects the factories just created are still in the identity map and the code under test may not query at all. Touching the relationships inside the block is what turns a lazy load into a counted query.

To make lazy loading fail outright instead of counting it, load with `raiseload("*")` in the code under test, or set `lazy="raise"` on relationships that should always be loaded explicitly. [Find and fix N+1 queries](https://flakytest.dev/guides/n-plus-one-queries/) explains `selectinload()` and `joinedload()`.

## Async: AsyncSession works the same way

`AsyncSession` passes its keyword arguments to the underlying `Session`, so the same pattern applies with an async connection. With pytest-asyncio 0.24 or later, keep the engine and the tests on one session-scoped event loop, because an async engine's connections belong to the loop that created them:

```python
# tests/conftest.py
import pytest_asyncio
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

from myapp.models import Base


@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def async_engine():
    engine = create_async_engine("postgresql+asyncpg://localhost/myapp_test")
    async with engine.begin() as connection:
        await connection.run_sync(Base.metadata.drop_all)
        await connection.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()


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

Mark the tests with `@pytest.mark.asyncio(loop_scope="session")` so they run on the same loop. `expire_on_commit=False` avoids attribute access after a commit triggering implicit IO, which raises in async code. Event listeners go on the sync engine: `event.listen(async_engine.sync_engine, "before_cursor_execute", ...)`. See [one event loop per async test is slow](https://flakytest.dev/guides/pytest-asyncio-event-loop-scope/) for the loop scope settings.

## Test the Alembic migrations

Building the test schema with `create_all()` is fast, but it means the tests never run your migrations. Two checks close that gap, and both belong in CI:

1. **Migrations run on an empty database.** Create a fresh database and run every migration:

   ```console
   $ alembic upgrade head
   ```

2. **Models match the migrations.** `alembic check` (Alembic 1.9 and later) runs the same comparison as `alembic revision --autogenerate` without writing a file, and exits with an error when it finds changes, so a model change without a migration fails the build:

   ```console
   $ alembic check
   ```

Run `alembic check` against the database you just upgraded. Autogenerate doesn't detect everything (a renamed table or column looks like a drop and an add, for example), so review generated migrations anyway.

If you'd rather have these as pytest tests, pytest-alembic ships them: `pytest --test-alembic` runs `test_single_head_revision`, `test_upgrade`, `test_model_definitions_match_ddl` and `test_up_down_consistency`. Keep them out of the fast local run if they're slow, and in CI always.

You can also build the test schema from the migrations instead of `create_all()`, with `command.upgrade(Config("alembic.ini"), "head")` from `alembic.command` and `alembic.config` in the session fixture. It's slower but tests exactly what production runs; keeping the database between runs makes it cheap again ([stop re-running migrations](https://flakytest.dev/guides/reuse-test-database/)).

## Find what's slow

`flakytest run` attributes test time to the database, schema creation, factories and connections, counts SQLAlchemy queries per test through engine events, and lists repeated queries. It suggests **Create the SQLAlchemy schema once, roll back per test** when `create_all` or `drop_all` runs in more than one test, and **Repeated queries (N+1) in the code under test** when a test runs the same query 10 or more times.

## Further reading

- [SQLAlchemy: joining a Session into an external transaction (such as for test suites)](https://docs.sqlalchemy.org/en/20/orm/session_transaction.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites)
- [SQLAlchemy: asyncio extension](https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html)
- [factory_boy: SQLAlchemy](https://factoryboy.readthedocs.io/en/stable/orms.html#sqlalchemy)
- [Alembic: autogenerate and alembic check](https://alembic.sqlalchemy.org/en/latest/autogenerate.html)
- [pytest-alembic](https://pytest-alembic.readthedocs.io/en/latest/)