Calling async SQLAlchemy from synchronous code

Synchronous code reaches an async engine in one of four ways — asyncio.run() for a one-shot script, a long-lived loop in a background thread for a server, run_sync() for sync code already inside an async call stack, or a second synchronous engine — and the right one depends on how many calls cross the boundary. This guide belongs to running concurrent queries with AsyncSession.

Quick Answer

For a script or a one-shot command, wrap the whole operation in one asyncio.run():

import asyncio

from sqlalchemy import select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from shop.models import Order


def recent_order_ids(limit: int = 20) -> list[int]:
    """Synchronous entry point for a CLI or a cron script."""

    async def run() -> list[int]:
        engine = create_async_engine("postgresql+asyncpg://shop:secret@db/shop")
        Session = async_sessionmaker(engine, expire_on_commit=False)
        try:
            async with Session() as session:
                result = await session.scalars(
                    select(Order.id).order_by(Order.placed_at.desc()).limit(limit)
                )
                return list(result)
        finally:
            await engine.dispose()

    return asyncio.run(run())

Two rules make this work. The engine is created and disposed inside the coroutine, because asyncio.run() closes the loop on return and a pooled connection would not survive it. And there is exactly one asyncio.run() per operation, not one per query — each call pays a full connection setup.

For a synchronous server making many calls, that cost is unacceptable; keep one loop alive in a thread instead, as the next section shows.

If the synchronous code is already running inside an async call stack — a callback, an event listener, a function passed to run_sync — do not reach for asyncio.run() at all. It raises RuntimeError: asyncio.run() cannot be called from a running event loop.

Execution Context & Async Workflow Integration

The boundary exists because an async driver needs a running event loop and synchronous code has none. Four bridges cross it, and the question that picks one is how many calls and how long does the process live.

One-shot: asyncio.run(). A migration script, a nightly job, a CLI command. The loop lives for one operation. Connection setup — TCP, TLS, authentication — is paid once, which against one operation's worth of queries is negligible.

Many calls in a long-lived process: a loop in a background thread. A Django or Flask application, a Celery worker, a synchronous gRPC server. Each call submits a coroutine to a loop that is already running, so the pool works normally and connection setup is paid once for the process. The caller blocks its own thread on the result, which is exactly what it was going to do anyway.

Sync code inside async: run_sync(). AsyncSession.run_sync() and AsyncConnection.run_sync() hand a synchronous Session or Connection to a function you supply, running it inside a greenlet. Blocking ORM calls inside that function work — including lazy loads — because when one needs I/O the greenlet suspends and the event loop awaits the driver on its behalf. This is how Base.metadata.create_all runs on an async connection, and it is the mechanism described in fixing GreenletSpawnError in async SQLAlchemy workflows.

Two engines. Nothing prevents one process from having a synchronous engine alongside the async one, pointing at the same database with a different driver:

from sqlalchemy import create_engine
from sqlalchemy.ext.asyncio import create_async_engine

async_engine = create_async_engine("postgresql+asyncpg://shop:secret@db/shop")
sync_engine = create_engine("postgresql+psycopg://shop:secret@db/shop", pool_size=5)

For an admin command, a report generator, or a synchronous background task, this is often the least surprising answer: no bridge, no loop management, no threading subtleties, and the ORM models are shared because mappings do not care which driver executes them. The cost is a second pool to size — both sets of connections count against the server's limit, as sizing pools behind PgBouncer and RDS Proxy explains.

Choosing between them is mostly about how much async code you already have. If the async engine exists because a FastAPI service uses it and a management command needs the same models, a synchronous engine is simpler than a bridge. If the data-access layer is async and shared, a bridge avoids duplicating it.

A Long-Lived Loop for a Synchronous Server

The background-loop bridge is short, and getting the lifecycle right matters more than the code:

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

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

T = TypeVar("T")


class AsyncDatabase:
    """One event loop, one engine, shared by synchronous callers."""

    def __init__(self, url: str, pool_size: int = 10) -> None:
        self._loop = asyncio.new_event_loop()
        self._ready = threading.Event()
        self._thread = threading.Thread(target=self._run, name="db-loop", daemon=True)
        self._thread.start()
        self._ready.wait(timeout=5)
        self._engine = create_async_engine(url, pool_size=pool_size)
        self._session_factory = async_sessionmaker(self._engine, expire_on_commit=False)

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

    def call(self, coro: Coroutine[Any, Any, T], timeout: float = 30.0) -> T:
        """Run a coroutine on the database loop and block until it finishes."""
        future = asyncio.run_coroutine_threadsafe(coro, self._loop)
        try:
            return future.result(timeout)
        except TimeoutError:
            future.cancel()
            raise

    def session(self) -> AsyncSession:
        return self._session_factory()

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

Callers write a coroutine and submit it:

from decimal import Decimal

from sqlalchemy.orm import selectinload

from shop.models import Order

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


def order_total(order_id: int) -> Decimal:
    async def run() -> Decimal:
        async with db.session() as session:
            order = await session.get(Order, order_id, options=[selectinload(Order.items)])
            return sum((item.price * item.quantity for item in order.items), Decimal("0"))
    return db.call(run())

Four details earn their keep.

The readiness event. Without it, run_coroutine_threadsafe can be called before run_forever() starts, and the coroutine is queued but never runs until the loop does — usually harmless, occasionally a hang at startup.

The engine is created after the thread starts, and disposed through call(). Both keep every connection's lifetime inside that one loop, which is the constraint in using one async engine across threads and event loops.

A timeout with cancellation. future.result(timeout) raising does not stop the coroutine; future.cancel() does. Without it, a slow query keeps a pooled connection busy after its caller has given up, and under load the pool drains — the exhaustion pattern in debugging QueuePool limit reached timeouts.

Each submitted coroutine opens its own session. One session shared between two submitted coroutines is concurrent use of a session, which fails the same way it would in ordinary async code.

One property of this design is easy to misread: the loop is not a concurrency limit. Twenty synchronous threads can each be blocked on future.result() while the loop interleaves their twenty queries, because the queries are I/O. The real limit is the pool. If pool_size + max_overflow is smaller than the number of calling threads, the surplus waits — which is correct behaviour, but it should be a deliberate choice rather than a surprise.

Django's async_to_sync (from asgiref) does roughly what call() does, with more care around context propagation, and is worth using where asgiref is already a dependency. It creates a new loop per call in the simple case, so the pooling caveat above still applies unless a loop is already running.

run_sync and the Greenlet Bridge

run_sync solves the opposite problem: synchronous code that must run where the connection is async. The function receives a real Session (or Connection) and may use any blocking API on it.

The canonical use is DDL, because MetaData.create_all has no async form:

async with engine.begin() as conn:
    await conn.run_sync(Base.metadata.create_all)

It is equally useful for reflection and for reusing existing synchronous helpers:

from sqlalchemy import inspect


def audit_schema(connection) -> dict[str, list[str]]:
    """Ordinary synchronous introspection, no await anywhere."""
    inspector = inspect(connection)
    return {
        table: [c["name"] for c in inspector.get_columns(table)]
        for table in inspector.get_table_names()
    }


async with engine.connect() as conn:
    schema = await conn.run_sync(audit_schema)

from decimal import Decimal

from sqlalchemy.orm import Session

from shop.models import Order


def apply_discount(session: Session, order_id: int) -> Decimal:
    """A pre-existing synchronous helper, reused unchanged."""
    order = session.get(Order, order_id)
    for item in order.items:            # a lazy load, inside a greenlet: fine
        item.price *= Decimal("0.9")
    session.flush()
    return sum((i.price * i.quantity for i in order.items), Decimal("0"))


async with async_session.begin():
    total = await async_session.run_sync(apply_discount, order_id)

Reflection of an existing database — Base.metadata.reflect, or automap — goes through the same call, as mapping classes to existing tables with reflection describes.

Two constraints apply inside a run_sync function. It must not call asyncio.run() or loop.run_until_complete() — the loop is running. And it must not block on something that is not database I/O: a time.sleep(), a synchronous HTTP call, or a lock held by a coroutine stalls the whole event loop, because the greenlet only yields on driver I/O.

That last point cuts the other way too. When synchronous, genuinely blocking work must happen inside an async request — a PDF render, a hashing round, a library with no async version — the tool is asyncio.to_thread, and the database work should stay outside it:

import asyncio

from sqlalchemy.orm import selectinload
from sqlalchemy.ext.asyncio import AsyncSession


async def build_invoice(session: AsyncSession, order_id: int) -> bytes:
    order = await session.get(Order, order_id, options=[selectinload(Order.items)])
    rows = [(i.sku, i.quantity, i.price) for i in order.items]   # plain data
    return await asyncio.to_thread(render_pdf, rows)             # no session inside

Passing the AsyncSession itself into to_thread would put it on another thread with no loop, which is the failure mode covered in fixing "another operation is in progress" errors with asyncpg. Extract what the blocking function needs first.

Resolving Warnings, Errors & Common Mistakes

Two mistakes deserve more than a row.

A loop per query rather than per operation. It is tempting to give every data-access function its own asyncio.run(), because each one is then independently callable. The result is a connection per function call: on a database 2 ms away, that is tens of milliseconds of setup for a 1 ms query, and a transaction cannot span two of them, so what looks like one unit of work is several. Wrap the operation — the whole command, the whole request — in one asyncio.run(), and let it call as many async functions as it likes.

Reaching for a bridge when a synchronous engine would do. If the synchronous caller is a Django management command, a Celery task, or an admin script, and it does not need to share async code, a create_engine() alongside the async one is less machinery and fewer failure modes. Bridges earn their complexity when the data-access layer is async and genuinely shared.

A last note on testing. Code that goes through a bridge is awkward to test through the bridge; it is much easier to test the async functions directly with pytest-asyncio and test the bridge itself once. The same reasoning applies to dependency overrides in overriding the database session dependency in FastAPI tests: keep the seam where the interesting code is.

Frequently Asked Questions

Related

• Running Concurrent Queries with AsyncSession — The parent guide: what one session can and cannot do.
• Using one async engine across threads and event loops — Why connections cannot move between loops.
• Fixing GreenletSpawnError in async SQLAlchemy workflows — The greenlet bridge from the ORM side.
• Disposing async engines on shutdown and in forked workers — Getting disposal ordering right.