Using one async engine across threads and event loops

An AsyncEngine object is safe to share, but the connections in its pool belong to the event loop that opened them — so a process with more than one loop needs either one engine per loop, disposal between loops, or NullPool so nothing is pooled at all. This guide belongs to running concurrent queries with AsyncSession.

Quick Answer

A pooled asyncpg connection holds transports and futures registered with the loop that created it. Reusing it on another loop fails.

Before — a module-level engine and a loop per operation:

import asyncio

from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine("postgresql+asyncpg://shop:secret@db/shop")


def count_orders() -> int:
    async def run() -> int:
        async with engine.connect() as conn:
            return await conn.scalar(text("SELECT count(*) FROM orders"))
    return asyncio.run(run())


count_orders()      # works: opens a connection on this loop, pools it
count_orders()      # RuntimeError: Task got Future attached to a different loop

After — an engine whose lifetime matches the loop's:

import asyncio

from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine

DATABASE_URL = "postgresql+asyncpg://shop:secret@db/shop"


def count_orders() -> int:
    async def run() -> int:
        engine = create_async_engine(DATABASE_URL)
        try:
            async with engine.connect() as conn:
                return await conn.scalar(text("SELECT count(*) FROM orders"))
        finally:
            await engine.dispose()          # inside the loop, before it closes
    return asyncio.run(run())

Or, when an engine must be shared across many such calls, keep the pool out of it:

from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.pool import NullPool

engine = create_async_engine(DATABASE_URL, poolclass=NullPool)
# Nothing is pooled, so no connection can be reused on a different loop.

A long-running service has one loop for its whole life and needs none of this. The problem only arises where a second loop exists.

Execution Context & Async Workflow Integration

asyncio ties I/O objects to a loop. When asyncpg connects, it creates a transport and awaits futures that belong to the running loop, and those references persist for the connection's life. SQLAlchemy's pool has no concept of loops: it stores the connection and hands it back on the next checkout, whichever loop is running then.

That combination produces RuntimeError: Task <...> got Future <...> attached to a different loop, and sometimes RuntimeError: Event loop is closed, depending on whether the original loop is merely different or already gone. Both mean the same thing.

Three process shapes create a second loop.

A test suite. pytest-asyncio's default is an event loop per test function, so a session-scoped engine fixture outlives many loops. The first test that touches the database works; the second fails. Matching the scopes fixes it — a session-scoped engine with asyncio_default_fixture_loop_scope = "session", or a function-scoped engine disposed at teardown, as in using pytest-asyncio fixtures with AsyncSession.

asyncio.run() per operation. A CLI command, a management script, or a synchronous framework that awaits one coroutine per request all create and destroy a loop each time. asyncio.run() closes the loop when it returns, so any connection left in the pool is dead.

Threads with their own loops. A thread pool where each worker runs asyncio.run() gives each thread a distinct loop, and a shared engine's pool will happily hand a connection created in one thread to another.

The engine object is thread-safe — its pool is synchronised, and sharing the object is fine. What is not shareable is the connection state inside it. So the rule is about lifetime rather than locking: an engine's pool must not outlive the loop that opened its connections.

Three arrangements satisfy it. One loop and one engine, which is the normal service shape. An engine created and disposed inside each loop, which suits per-operation loops. Or NullPool, which opens a connection per checkout and closes it on return, so nothing survives to be misused — the same reasoning that makes it right for serverless, in using NullPool for serverless and AWS Lambda.

engine.dispose() is the tool for the middle option, and it must be awaited inside the loop — after asyncio.run() returns, there is no loop to close connections on, which is the disposal ordering described in disposing async engines on shutdown and in forked workers.

Resolving Warnings, Errors & Common Mistakes

The fourth row is worth separating from the rest because it looks related and is not. asyncio.run() cannot be nested, so calling it from inside async code fails immediately with a clear message — the situation covered in calling async SQLAlchemy from synchronous code.

For a CLI with many commands, creating an engine per command is wasteful if the commands are short. A small helper keeps one loop for the process and lets every command share one engine on it:

import asyncio
from collections.abc import Awaitable, Callable
from typing import TypeVar

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

T = TypeVar("T")

_loop = asyncio.new_event_loop()
engine = create_async_engine(DATABASE_URL)
Session = async_sessionmaker(engine, expire_on_commit=False)


def run(coro: Awaitable[T]) -> T:
    """Run a coroutine on the process's single loop."""
    return _loop.run_until_complete(coro)


def shutdown() -> None:
    run(engine.dispose())
    _loop.close()

Every command calls run(...), so there is exactly one loop and the pool stays valid for the process's life. Registering shutdown() with atexit — or calling it at the end of main() — closes the connections while the loop is still alive.

For threads, the equivalent is one engine per thread, which a thread-local makes tidy:

import asyncio
import threading

from sqlalchemy.ext.asyncio import create_async_engine

_local = threading.local()


def thread_engine():
    engine = getattr(_local, "engine", None)
    if engine is None:
        engine = create_async_engine(DATABASE_URL, pool_size=2, max_overflow=0)
        _local.engine = engine
    return engine

Each thread's engine is used only by that thread's loop. Note the small pool: with several threads, the totals multiply, which is the same arithmetic as for replicas.

Advanced: Mixing Threads and Loops Deliberately

Sometimes a process genuinely needs both: a synchronous framework serving requests in threads, with async database access, or an async service offloading a blocking library. Two arrangements work, and the choice is about where the boundary sits.

One loop in a dedicated thread, shared by everyone. A background thread runs the loop, and synchronous callers submit coroutines to it with asyncio.run_coroutine_threadsafe. There is exactly one loop and one engine, so the pool is always valid:

import asyncio
import threading
from collections.abc import Coroutine
from typing import Any, TypeVar

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

T = TypeVar("T")


class AsyncBridge:
    """One event loop in one thread, shared by synchronous callers."""

    def __init__(self, url: str) -> None:
        self._loop = asyncio.new_event_loop()
        self._thread = threading.Thread(target=self._run, daemon=True, name="db-loop")
        self._thread.start()
        self.engine = create_async_engine(url, pool_size=10)
        self.Session = async_sessionmaker(self.engine, expire_on_commit=False)

    def _run(self) -> None:
        asyncio.set_event_loop(self._loop)
        self._loop.run_forever()

    def call(self, coro: Coroutine[Any, Any, T], timeout: float = 30.0) -> T:
        future = asyncio.run_coroutine_threadsafe(coro, self._loop)
        return future.result(timeout)

    def close(self) -> None:
        self.call(self.engine.dispose())
        self._loop.call_soon_threadsafe(self._loop.stop)
        self._thread.join(timeout=5)

Every connection is created on that one loop, so nothing can cross a boundary. The cost is that all database work is serialised through one loop's scheduling — which is fine, because the work is I/O-bound and the loop is not the bottleneck — and that each synchronous caller blocks its own thread waiting for future.result().

A loop per thread, an engine per thread. Simpler to reason about, and it multiplies pools: with eight worker threads and pool_size=2 each, that is sixteen connections. Right when threads are long-lived and few; wasteful when they are many or short-lived.

Two details matter for either arrangement.

Pool arithmetic changes. Connections are per engine, and engines are per loop, so the total is engines times pool size. Under a connection-limited database that number is the one that counts, and it is easy to overshoot — the reason sizing pools behind PgBouncer and RDS Proxy recommends counting every process and thread.

Sessions still belong to one task. Nothing about threading changes the rule that an AsyncSession is used by one task at a time. A bridge that hands the same session to two synchronous callers has the problem described in fixing "another operation is in progress" errors with asyncpg, with a thread boundary making it harder to see. Each submitted coroutine should open its own session.

Getting the Test Suite Right

Test suites hit this more than production code does, because a suite deliberately creates and destroys loops. Two arrangements work; mixing them is what fails.

A session-scoped engine with a session-scoped loop. The fastest option: the schema is created once, the pool is warm, and every test shares the loop.

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

import pytest_asyncio
from sqlalchemy.ext.asyncio import create_async_engine

from shop.models import Base


@pytest_asyncio.fixture(scope="session")
async def engine(postgres_url: str):
    engine = create_async_engine(postgres_url)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()

The loop-scope setting is the part people miss: without it, pytest-asyncio uses a function-scoped loop and the session-scoped engine is exactly the mismatch this guide is about.

A function-scoped engine, disposed at teardown. Slower — a new pool per test, and create_all per test unless the schema is created separately — and fully isolated, which suits a suite where tests alter the schema:

import pytest_asyncio
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.pool import NullPool


@pytest_asyncio.fixture
async def engine(postgres_url: str):
    engine = create_async_engine(postgres_url, poolclass=NullPool)
    yield engine
    await engine.dispose()

NullPool here removes even the possibility of a stale pooled connection, at the cost of a connection per checkout — negligible against a local database.

One failure worth recognising: a suite that passes when run with -k on a single test and fails on the full run. That is almost always this problem, because the second database-touching test is the first to inherit a connection from a closed loop. Running a single test never reaches it.

Finally, the application's own engine deserves care in tests. A module-level engine created at import time is bound to no loop until first use, and then to whichever loop used it first — which in a test suite is the first test. Overriding the application's session dependency with a test-managed factory, as in overriding the database session dependency in FastAPI tests, keeps the application engine out of the tests entirely — which is the cleanest way to sidestep the whole question.

Frequently Asked Questions

Related

• Running Concurrent Queries with AsyncSession — The parent guide: what can be shared between tasks.
• Calling async SQLAlchemy from synchronous code — Crossing the boundary from the other direction.
• Disposing async engines on shutdown and in forked workers — Disposal ordering, and forks.
• Using pytest-asyncio fixtures with AsyncSession — Loop scopes in a test suite.