A minimal, production-style example of testing a FastAPI application that uses async SQLAlchemy without running into RuntimeError: Task attached to a different loop.
When you start writing integration tests for a FastAPI app backed by async SQLAlchemy, things tend to go wrong in a specific and frustrating way, the error says something about a task being "attached to a different loop."
This is not a bug in your application code. It is a misunderstanding of how async database connection pools behave across event loops and the fix is not obvious from the SQLAlchemy or pytest-asyncio documentation alone. This repo shows the problem concretely, explains what causes it, and demonstrates a test setup that handles it correctly.
app/
├── api.py # Route handlers
├── db.py # Engine and sessionmaker factory
├── dependencies.py # get_db, verify_auth
├── main.py # create_app()
├── models.py
└── tests/
├── conftest.py
└── integration/
└── test_users.py
[pytest]
asyncio_mode = auto
asyncio_default_test_loop_scope = session
asyncio_default_fixture_loop_scope = sessionasyncio_default_test_loop_scope = session and asyncio_default_fixture_loop_scope = session are the
key settings they tell pytest-asyncio to reuse one event loop for the whole session instead of
creating one per test.
Session-scoped engine created once, pool stays on one loop, disposed at the end.
@pytest.fixture(scope="session")
async def engine():
engine = create_async_engine(DATABASE_URL, pool_pre_ping=True)
yield engine
await engine.dispose()Migrations Alembic's upgrade is synchronous, so it runs in a thread to avoid blocking the loop.
@pytest.fixture(scope="session", autouse=True)
async def migrations(engine):
await asyncio.to_thread(lambda: upgrade(AlembicConfig("alembic.ini"), "head"))
yieldTransactional isolation
Each test gets a fresh AsyncSession that is bound to a real database connection opened inside a transaction. After the test finishes regardless of whether it passed or failed both the session and the outer transaction are rolled back. This means:
- Tests are isolated from each other.
- The database is left in a clean state without needing to truncate tables between tests.
join_transaction_mode="create_savepoint"allowsdb.commit()calls inside route handlers to work without breaking the outer transaction.
@pytest.fixture
async def db_session(engine):
connection = await engine.connect()
transaction = await connection.begin()
session = AsyncSession(bind=connection, expire_on_commit=False,
join_transaction_mode="create_savepoint")
try:
yield session
finally:
await session.close()
await transaction.rollback()
await connection.close()Dependency overrides injects the test session and bypasses auth.
@pytest.fixture
def test_app(db_session):
app = create_app()
app.dependency_overrides[get_db] = lambda: (yield db_session)
app.dependency_overrides[verify_auth] = lambda: "test"
yield app
app.dependency_overrides.clear()HTTP client talks to the app in-process, no network port required.
@pytest.fixture
async def client(test_app):
async with AsyncClient(transport=ASGITransport(app=test_app),
base_url="http://test") as ac:
yield acFunction-scoped engine the pool binds to loop A, loop A closes, loop B starts, error fires. This is the most common mistake.
Sharing the app's sessionmaker in tests get_engine() / get_sessionmaker() initialize lazily. If they run during tests they create a pool on whichever loop happens to be active, which may not be the test loop.
# set connection URL
export TEST_DB_URL=postgresql+asyncpg://user:password@localhost/test_db
uv sync --dev
alembic upgrade head
pytest