Running tests against a PostgreSQL testcontainer

Start one PostgreSQL container per test session, run the real migrations into it once, and hand every worker a database cloned from that template — the container costs a few seconds at the start of the run and buys you concurrency, native types, deferrable constraints and isolation levels that no SQLite suite can reproduce. This is the fidelity half of testing async SQLAlchemy applications.

Quick Answer

One container, one migration run, N cloned databases A five-stage session. The container starts once and the run waits for the server to accept connections. Migrations run once into a template database, which is the slow step and the one worth doing exactly once. Each parallel worker then creates its own database with CREATE DATABASE ... TEMPLATE, which copies the template at the file level and takes well under a second regardless of schema size. Tests run against those clones with ordinary rollback isolation. At the end the container is destroyed, taking every database with it — so there is no cleanup step that can fail and leave residue for the next run. container starts, once per session wait for the server to accept connections seconds, not minutes migrations run into a template database the real alembic upgrade head the slow step — done once each worker clones the template CREATE DATABASE ... TEMPLATE a file-level copy, sub-second tests run with rollback isolation per test, as usual unchanged by any of this container destroyed at session end every database goes with it Templating is what makes parallel runs cheap: migrating once and cloning N times costs the migration once, where migrating per worker multiplies it by the worker count.

Before — a shared test database somebody has to maintain:

# Every developer needs a local PostgreSQL at a known name, with the schema
# up to date, and CI needs a service container configured separately.
TEST_URL = "postgresql+asyncpg://app:secret@localhost:5432/orders_test"

After — a disposable container the test session owns:

# conftest.py
from __future__ import annotations

from collections.abc import AsyncIterator, Iterator

import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
from testcontainers.postgres import PostgresContainer


@pytest.fixture(scope="session")
def postgres() -> Iterator[PostgresContainer]:
    with PostgresContainer("postgres:16-alpine") as container:
        yield container


@pytest_asyncio.fixture(scope="session")
async def engine(postgres: PostgresContainer) -> AsyncIterator[AsyncEngine]:
    url = postgres.get_connection_url().replace("psycopg2", "asyncpg")
    engine = create_async_engine(url)
    await _upgrade_to_head(url)          # the real migrations, not create_all
    yield engine
    await engine.dispose()

Nothing outside the test session has to exist: no local server at a known port, no CI service definition, no schema anyone has to remember to update. The .replace("psycopg2", "asyncpg") is the one wart — testcontainers returns a synchronous URL, and SQLAlchemy needs the async driver named in it.

Execution Context & Async Workflow Integration

The container is synchronous infrastructure serving an async suite, and the two need to meet in the right order. The container fixture is an ordinary @pytest.fixture because starting Docker involves no await; the engine fixture is a pytest_asyncio.fixture because create_async_engine and dispose() do. Ordering falls out of the dependency: engine requests postgres, so the container is up before any connection is attempted.

Running the migrations is the interesting part, because Alembic's API is synchronous while the rest of the fixture is not. Blocking the loop inside a session-scoped fixture is harmless — nothing else is running — but doing it in a thread keeps the pattern honest and matches what an application start-up would do.

import asyncio


async def _upgrade_to_head(url: str) -> None:
    """Run the real migrations, so the suite tests the schema migrations produce."""
    from alembic import command
    from alembic.config import Config

    config = Config("alembic.ini")
    # Alembic's own engine is synchronous; give it the sync URL.
    config.set_main_option("sqlalchemy.url", url.replace("asyncpg", "psycopg2"))
    await asyncio.to_thread(command.upgrade, config, "head")

Using migrations rather than Base.metadata.create_all() is the whole point of having a real database in the loop. create_all builds the schema the models describe; migrations build the schema production will actually have. Those diverge exactly when a migration is wrong — a missing index, a column left nullable, an enum value never added — and a suite built on create_all cannot see any of it. The review discipline that catches those before they ship is in autogenerating and reviewing migration scripts; running them here is the second line of defence.

Waiting for readiness deserves one note. testcontainers waits for the port to accept connections, which happens slightly before PostgreSQL is ready to serve queries. The library handles this for the standard image, but a custom image with an init script needs an explicit readiness probe — a SELECT 1 in a retry loop — or the first migration occasionally fails with a connection reset.

What the container buys that SQLite cannot Four categories. Real write concurrency means two transactions can genuinely conflict, so deadlock handling and serialisation-failure retries are exercised rather than assumed. Native types mean JSONB, ARRAY and enum behave as they will in production, including the operators and index types that only exist for them. Deferrable constraints mean a test can verify that a check really is deferred to commit, which SQLite cannot express at all. And isolation levels mean REPEATABLE READ and SERIALIZABLE behave as they will in production, including producing the serialisation failures a retry loop exists to handle. write concurrency two transactions really conflict deadlocks and 40001 retries SQLite serialises writers native types JSONB, ARRAY, enum with their operators and index types SQLite degrades them to text deferrable constraints checked at COMMIT, not per statement the expand-contract pattern needs this SQLite has no equivalent isolation levels REPEATABLE READ, SERIALIZABLE and the 40001 they produce SQLite has no MVCC Every one of these is a place where a SQLite suite reports success for code that will fail in production — which is the argument for having both suites, not either.

Resolving Warnings, Errors & Common Mistakes

Error / WarningRoot CauseProduction Fix
docker.errors.DockerException: Error while fetching server API versionNo Docker daemon reachable — common in CI runners without a socket mounted, and on rootless setups.Point DOCKER_HOST at the right socket, or skip the marked tests when Docker is absent rather than failing the run.
sqlalchemy.exc.InvalidRequestError: The asyncio extension requires an async driverget_connection_url() returned a psycopg2 URL.Replace the driver in the URL before passing it to create_async_engine.
ConnectionResetError on the first queryThe port accepted before the server was ready.Add an explicit readiness probe: SELECT 1 in a short retry loop before migrating.
Suite is slow, migrations dominateMigrations run per worker instead of once.Migrate a template database once, then CREATE DATABASE ... TEMPLATE per worker.
psycopg2.errors.ObjectInUse: source database is being accessed by other usersCloning from a template while a connection to it is open.Dispose the template engine before cloning, and never connect to the template during the run.
Container left running after a crashed runThe session fixture never reached its teardown.Label containers and prune by label in CI; enable reuse deliberately rather than by accident locally.

Advanced: Template Databases and Parallel Workers

The container is cheap; the migration run is not. On a mature schema, alembic upgrade head can take tens of seconds, and multiplying that by eight parallel workers turns a fast suite into a slow one. PostgreSQL's template mechanism removes the multiplication entirely.

# conftest.py — migrate once, clone per worker
from __future__ import annotations

import os
from collections.abc import AsyncIterator

import pytest_asyncio
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine


@pytest_asyncio.fixture(scope="session")
async def template_url(postgres) -> str:
    """Build the schema once, in a database nothing else will connect to."""
    admin_url = postgres.get_connection_url().replace("psycopg2", "asyncpg")
    admin = create_async_engine(admin_url, isolation_level="AUTOCOMMIT")
    async with admin.connect() as connection:
        await connection.execute(text("DROP DATABASE IF EXISTS tmpl"))
        await connection.execute(text("CREATE DATABASE tmpl"))
    await admin.dispose()

    tmpl_url = admin_url.rsplit("/", 1)[0] + "/tmpl"
    await _upgrade_to_head(tmpl_url)
    return tmpl_url


@pytest_asyncio.fixture(scope="session")
async def engine(postgres, template_url: str) -> AsyncIterator[AsyncEngine]:
    """One database per xdist worker, cloned from the migrated template."""
    worker = os.environ.get("PYTEST_XDIST_WORKER", "gw0")
    admin_url = postgres.get_connection_url().replace("psycopg2", "asyncpg")
    admin = create_async_engine(admin_url, isolation_level="AUTOCOMMIT")
    async with admin.connect() as connection:
        await connection.execute(text(f'DROP DATABASE IF EXISTS "test_{worker}"'))
        await connection.execute(text(f'CREATE DATABASE "test_{worker}" TEMPLATE tmpl'))
    await admin.dispose()

    engine = create_async_engine(admin_url.rsplit("/", 1)[0] + f"/test_{worker}")
    yield engine
    await engine.dispose()

Three details are load-bearing. CREATE DATABASE cannot run inside a transaction, hence isolation_level="AUTOCOMMIT" on the admin engine. Nothing may hold a connection to the template while it is being cloned, which is why the template engine is disposed before any worker starts — the error otherwise is source database is being accessed by other users, and it appears only under parallelism. And each admin engine is disposed immediately, because an admin connection left open is one more connection against the container's max_connections.

The payoff is that adding workers costs a sub-second clone each rather than a full migration each. On a suite with a minute of migrations and eight workers, that is the difference between eight minutes of setup and one.

Keeping the container off the critical path Three techniques. A pytest marker separates the tests that genuinely need PostgreSQL from the majority that do not, so the default local run stays on SQLite and finishes in seconds. Container reuse — enabled with a flag and a label — leaves the container running between local invocations, so the start-up cost is paid once a day rather than once a run. And building the schema once into a template, then cloning per worker, keeps the migration cost from multiplying with parallelism. Together these mean the container-backed suite is something CI runs on every push and a developer runs before pushing, rather than something everyone avoids. mark the tests that need it @pytest.mark.postgres, deselected by default the fast suite stays on SQLite and finishes in seconds reuse the container between local runs testcontainers reuse, keyed on a label start-up paid once a day rather than once a run migrate once, clone per worker CREATE DATABASE ... TEMPLATE the migration cost stops scaling with the worker count The goal is not to make the container-backed suite fast enough to run constantly, but cheap enough that nobody is tempted to skip it before pushing.

Deciding Which Tests Need the Container

A container-backed suite that runs on every save is a suite people stop running. The split that works is a marker plus a default deselection.

# pyproject.toml
[tool.pytest.ini_options]
markers = ["postgres: needs a real PostgreSQL, not SQLite"]
addopts = "-m 'not postgres'"        # the default run is the fast one

pytest                 # fast: SQLite only, seconds
pytest -m postgres     # the fidelity suite, before pushing
pytest -m ''           # everything, in CI

What belongs behind the marker is anything whose subject is the database. Concurrency tests, obviously — two transactions contending for a row, a deadlock, a SERIALIZABLE retry loop. Anything using PostgreSQL-specific SQL: ON CONFLICT, JSONB operators, window functions with unusual frames, partial or expression indexes. Anything about constraint timing, including the deferred-check step of an expand-and-contract migration. And the migration run itself.

What does not belong behind it is the bulk of a service layer: validation, mapping, query construction, business rules. Those are about Python, and running them against SQLite in memory keeps the feedback loop tight enough that people actually use it.

The one habit that keeps the split honest is to move a test across the line when it earns it. A test that starts failing intermittently on SQLite because it grew a concurrency assumption belongs behind the marker; a container-backed test that turns out to assert nothing PostgreSQL-specific belongs back in the fast suite. Reviewing that boundary occasionally costs little and keeps both suites meaning what they claim to.

Frequently Asked Questions

Is a testcontainer slower than a service container in CI?

Slightly, because the image is pulled and started by the test session rather than by the CI runner in parallel with checkout. In exchange the same command works identically on a laptop and in CI, and there is no second place where the database version and configuration are declared — which is usually worth more than the few seconds.

Can I reuse a container between local runs?

Yes. testcontainers supports reuse when enabled in configuration and the container carries a stable label; the session then attaches to a running container instead of starting one. Keep it opt-in for local use only — a reused container in CI would carry state between builds, which is exactly what the container was meant to prevent.

Should the container run migrations or create_all?

Migrations. Building the schema from model metadata tests the schema your models describe, which is the one thing you already know is consistent. Running the migrations tests the schema production will have, and is the only way a wrong migration is caught before it reaches an environment that matters.

How many connections does a parallel run need?

Roughly one per worker for the test transaction, plus whatever each worker's pool keeps open, plus the admin connections during setup. Eight workers is comfortably under the default max_connections of 100; sixty-four workers is not, and the arithmetic is the same as for an application fleet.