Using pytest-asyncio fixtures with AsyncSession

Set asyncio_mode = "auto" and asyncio_default_fixture_loop_scope = "session", then give the engine fixture the same scope as the loop — almost every "attached to a different loop" failure in an async SQLAlchemy suite is a connection outliving the loop that created it. This is the fixture-plumbing companion to testing async SQLAlchemy applications.

Quick Answer

Before — a hand-written loop fixture and mismatched scopes:

# conftest.py — the shape that produces "attached to a different loop"
import asyncio

import pytest


@pytest.fixture(scope="session")
def event_loop():                       # deprecated, and the source of the problem
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()


engine = create_async_engine(TEST_URL)  # created at import, before any loop exists

After — configuration instead of fixtures:

# pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "session"

# conftest.py — no event_loop fixture at all
from collections.abc import AsyncIterator

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

from myapp.models import Base


@pytest_asyncio.fixture(scope="session")
async def engine() -> AsyncIterator[AsyncEngine]:
    engine = create_async_engine("postgresql+asyncpg://app:secret@localhost/orders_test")
    async with engine.begin() as connection:
        await connection.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()

Two changes carry it. The hand-written event_loop fixture is gone — overriding it has been deprecated, and the configuration option replaces it. And the engine moved from module scope into a fixture, so it is created after the loop exists rather than before.

Execution Context & Async Workflow Integration

An asyncio connection is bound to the loop it was created on: its socket is registered with that loop's selector, and its futures are that loop's futures. Handing it to a different loop is not a SQLAlchemy limitation but an asyncio one, and the error message — Task got Future attached to a different loop — names the mechanism rather than the cause.

That gives a single rule: the loop must be at least as long-lived as anything holding a connection. A session-scoped engine with a function-scoped loop violates it, and does so intermittently — the first test creates connections on loop A, the pool keeps them, and the second test on loop B fails when it reuses one.

There are two ways to satisfy the rule and they are both legitimate:

# Option 1 — a long-lived loop, so the pool is genuinely reused (preferred)
# pyproject.toml: asyncio_default_fixture_loop_scope = "session"
@pytest_asyncio.fixture(scope="session")
async def engine() -> AsyncIterator[AsyncEngine]:
    engine = create_async_engine(TEST_URL)
    yield engine
    await engine.dispose()

# Option 2 — a per-test loop, so the engine must be per-test and must not pool
from sqlalchemy.pool import NullPool


@pytest_asyncio.fixture
async def engine() -> AsyncIterator[AsyncEngine]:
    engine = create_async_engine(TEST_URL, poolclass=NullPool)
    yield engine
    await engine.dispose()

Option 1 is faster because the pool survives across tests. Option 2 is simpler to reason about and is worth reaching for when a suite has some intractable per-test loop requirement — but NullPool is not optional there: without it, the engine's pool holds connections from the previous test's loop and the failure returns.

The same reasoning explains why an engine must not be created at import time. Module-level code runs during collection, before the plugin has installed any loop, so the engine's pool is created against whatever loop happens to exist — often none. This is the same failure the Celery integration guide describes for prefork workers: an engine that predates the execution context it will be used from.

Resolving Warnings, Errors & Common Mistakes

Advanced: Fixture Layering for a Real Suite

A suite of any size ends up with more than three fixtures, and the layering is worth designing once rather than growing by accretion.

# conftest.py — the full chain, with each layer doing exactly one thing
from __future__ import annotations

import os
from collections.abc import AsyncIterator

import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import (
    AsyncConnection, AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine,
)

from myapp.models import Base


@pytest_asyncio.fixture(scope="session")
async def engine() -> AsyncIterator[AsyncEngine]:
    """One engine, one pool, one schema build for the whole run."""
    worker = os.environ.get("PYTEST_XDIST_WORKER", "gw0")
    url = f"postgresql+asyncpg://app:secret@localhost/orders_test_{worker}"
    engine = create_async_engine(url)
    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
async def connection(engine: AsyncEngine) -> AsyncIterator[AsyncConnection]:
    """One never-committed transaction per test."""
    async with engine.connect() as connection:
        transaction = await connection.begin()
        yield connection
        await transaction.rollback()


@pytest_asyncio.fixture
async def session(connection: AsyncConnection) -> AsyncIterator[AsyncSession]:
    """The session every test and every service under test shares."""
    factory = async_sessionmaker(
        bind=connection, expire_on_commit=False,
        join_transaction_mode="create_savepoint",
    )
    async with factory() as session:
        yield session


@pytest_asyncio.fixture
async def seeded(session: AsyncSession) -> AsyncSession:
    """Reference data most tests want, inside the same transaction."""
    session.add_all([Currency(code="GBP"), Currency(code="USD")])
    await session.flush()
    return session

Three properties make this layering hold up. Each fixture has exactly one responsibility, so a change to isolation strategy touches one function. Fixtures depend only outward — function scope may depend on session scope, never the reverse — which is what ScopeMismatch enforces. And the seed fixture builds its data inside the test transaction rather than in a session-scoped one, so seeded rows roll back with everything else and a test that mutates them cannot affect the next.

The one thing deliberately absent is an autouse fixture. An autouse=True database fixture makes every test in the suite — including pure unit tests with no database involvement — pay for a connection checkout. Requesting session explicitly keeps that cost visible and keeps the fast tests fast.

Running the Suite in Parallel

Rollback isolation makes tests independent, which makes pytest -n auto safe as soon as each worker has its own database. The PYTEST_XDIST_WORKER variable in the engine fixture above is the whole mechanism; the databases themselves are created once, before the run.

# create one database per worker, once, before the suite runs
for i in $(seq 0 7); do
  createdb "orders_test_gw${i}" 2>/dev/null || true
done
pytest -n 8

Two considerations decide whether this is worth doing. Schema setup happens once per worker rather than once per run, so a slow create_all or a full migration run is multiplied by the worker count — with migrations that can dominate, and the usual answer is to build one template database and createdb --template from it, which is close to instantaneous.

And the database server itself becomes a shared resource: eight workers each holding an open transaction is eight connections plus whatever the pool keeps, which is trivial for a local PostgreSQL and worth checking against max_connections in CI. The arithmetic is the same as for an application fleet, and the reasoning in tuning connection pools for cloud databases applies unchanged — a test run is just another client with a connection budget.

The failure mode to watch for is a test that passes serially and fails under -n, which nearly always means shared state that is not the database: a temporary file at a fixed path, a module-level cache, a fixed port. Those are worth fixing rather than working around, because the same shared state will eventually cause a flake in the serial suite too.

Frequently Asked Questions

Related

• Testing Async SQLAlchemy Applications — The parent guide: fixture layout, the rollback pattern, and the SQLite-versus-PostgreSQL split.
• Rolling back database state between async tests — The nested-transaction fixture these scopes exist to support.
• Running tests against a PostgreSQL testcontainer — Starting the database the session-scoped engine connects to.
• Using SQLAlchemy async with Celery task workers — The same "engine created before its execution context" bug, in a prefork worker.