Running Concurrent Queries with AsyncSession

Share the AsyncEngine and the async_sessionmaker across the whole process, and give every concurrent task its own AsyncSession — a session is a single unit of work on a single connection, and asyncpg allows one statement per connection at a time. This topic, part of async engines, dialects and connection pooling, covers how to fan queries out safely, bound that fan-out by the pool, and keep session state from leaking across task boundaries.

Concept & Execution Model

SQLAlchemy's async layer is a thin, faithful wrapper over the synchronous ORM, and it inherits the synchronous ORM's ownership model exactly. An AsyncEngine owns a pool and is meant to be shared by everything in the process. An async_sessionmaker is a configured factory, also shared. An AsyncSession is a unit of work — an identity map, a set of pending changes and one transaction on one connection — and it belongs to a single task at a time. The connection beneath it, an asyncpg protocol object, can execute exactly one statement at a time.

What can be shared between tasks Four bands from the top. The AsyncEngine and its pool are process-wide and designed to be shared by every task. The async_sessionmaker is a factory, also shared. An AsyncSession is a unit of work with one transaction and must belong to one task at a time. The asyncpg connection underneath a session runs one statement at a time and is never shared concurrently. AsyncEngine + pool — share freely one per process and database; hands out connections to whoever asks async_sessionmaker — share freely a factory with configuration; calling it makes a new, independent session AsyncSession — one task at a time identity map, pending changes, one transaction: a single unit of work asyncpg connection — one statement at a time a single protocol conversation; overlapping calls raise InterfaceError Concurrency is created at the factory line: many sessions, each owned by one task.

Concurrency therefore lives at the factory line. The pattern is always the same: a shared factory, and each concurrent unit of work calling it to get a session of its own.

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

# Shared by the whole process.
engine = create_async_engine(
    "postgresql+asyncpg://shop:secret@db/shop",
    pool_size=10,
    max_overflow=5,
    pool_pre_ping=True,
)
Session = async_sessionmaker(engine, expire_on_commit=False)


async def reserve_stock(product_id: int, quantity: int) -> None:
    # Owned by this task alone, from here to the end of the block.
    async with Session.begin() as session:
        ...

The distinction matters more under asyncio than it ever did under threads, because async code looks sequential. await session.execute(q1) followed by await session.execute(q2) reads like two steps, and is. The same two calls wrapped in asyncio.gather() read almost identically and are two tasks racing for one connection — which asyncpg rejects with InterfaceError: cannot perform operation: another operation is in progress. That error, and the three other exceptions that describe the same mistake, are worked through in fixing "another operation is in progress" errors.

It helps to be precise about what "concurrent" means. Tasks are concurrent when their awaits overlap in time — when one is suspended waiting for the database while another runs. Passing a session from one task to another sequentially, where the first has finished all its awaits before the second starts, is allowed. Code that is only safe because the awaits happen not to overlap on a fast local database is not.

This topic sits within async engines, dialects and connection pooling, and it depends directly on two neighbours: how the pool is sized, because every concurrent session wants a connection, and how sessions are scoped, which the AsyncSession lifecycle covers in depth.

Query Construction & Async Execution Patterns

The statements themselves do not change when they run concurrently — select(), insert() and update() are built the same way — but the execution scaffolding around them does. Two shapes cover almost every case: fanning out independent reads, and processing a queue of items with bounded concurrency.

The same rule in both worlds Left, synchronous: a ThreadPoolExecutor submits work, each worker thread calls Session() to get its own session, and the engine pool hands each its own connection. Right, asynchronous: a TaskGroup creates tasks, each task calls the async session factory to get its own AsyncSession, and the async pool hands each its own connection. The ownership rule, one session per unit of concurrency, is identical. sync — threads ThreadPoolExecutor(max_workers=4) each worker: with Session() as s Session is not thread-safe pool_size bounds real parallelism async — tasks asyncio.TaskGroup() + Semaphore(4) each task: async with Session() as s AsyncSession is not task-safe pool_size bounds real parallelism Moving from threads to tasks changes the scheduling, not the ownership rule.

The synchronous version of a bounded fan-out uses a thread pool, with a sessionmaker producing a session per worker:

# Sync — one Session per worker thread
from concurrent.futures import ThreadPoolExecutor

from sqlalchemy import create_engine, select
from sqlalchemy.orm import sessionmaker

from shop.models import Product

engine = create_engine("postgresql+psycopg://shop:secret@db/shop", pool_size=4)
Session = sessionmaker(engine, expire_on_commit=False)


def price_of(sku: str) -> tuple[str, int | None]:
    with Session() as session:
        return sku, session.scalar(select(Product.price_cents).where(Product.sku == sku))


def price_many(skus: list[str]) -> dict[str, int | None]:
    with ThreadPoolExecutor(max_workers=4) as pool:
        return dict(pool.map(price_of, skus))

The asynchronous version replaces the executor with a TaskGroup and the worker count with a semaphore, and keeps everything else:

# Async — one AsyncSession per task
import asyncio

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

from shop.models import Product

engine = create_async_engine("postgresql+asyncpg://shop:secret@db/shop", pool_size=4)
Session = async_sessionmaker(engine, expire_on_commit=False)


async def price_of(sku: str, limit: asyncio.Semaphore) -> tuple[str, int | None]:
    async with limit:
        async with Session() as session:
            return sku, await session.scalar(
                select(Product.price_cents).where(Product.sku == sku)
            )


async def price_many(skus: list[str]) -> dict[str, int | None]:
    limit = asyncio.Semaphore(4)
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(price_of(sku, limit)) for sku in skus]
    return dict(task.result() for task in tasks)

Two details carry over from threads and are easy to lose. The concurrency limit should be at or below pool_size, because tasks beyond that wait in the pool queue, where waiting is invisible and bounded only by pool_timeout. And the semaphore is acquired before the session is opened, so a waiting task holds nothing.

For work of a fixed, small shape — three aggregates for a dashboard, a lookup in two databases — a semaphore is unnecessary and the TaskGroup alone is right. Running queries in parallel with asyncio.TaskGroup covers that shape, including except* handling, timeouts and what happens to a query when its task is cancelled.

Before either, check that the example above should not simply be one query. select(Product.sku, Product.price_cents).where(Product.sku.in_(skus)) answers price_many() in one round trip on one connection, and for any list that fits in a statement it will beat both versions.

State Management & Session Boundaries

A session's identity map is keyed per session, so two tasks with two sessions that load the same Order row get two distinct Python objects. That is the correct behaviour — each unit of work has its own view — but it has consequences that code written for one session does not anticipate.

Session boundaries across a request Five steps. A request arrives and the dependency opens a session. The handler reads and writes through it and commits. It schedules follow-up work, passing only primary keys, not ORM objects or the session. The response is sent and the request session closes. The background task opens its own session, reloads the rows it needs by key, and commits separately. request arrives dependency opens session A yield session handler works reads, writes, commits on A passes order.id onward spawn follow-up work argument: an integer key never the session or an ORM object response sent session A closes connection back in the pool background task opens session B, reloads by key Passing keys instead of objects is what lets two sessions own the same row at different times.

Objects do not cross task boundaries safely. An Order loaded in one task's session is attached to that session. Handed to another task, it is either still attached to a session the other task does not own, or detached once the first session closes. Touching an unloaded attribute on it raises DetachedInstanceError, or under async MissingGreenlet. Pass primary keys between tasks and reload, or pass plain data — a dataclass or a dictionary built from the loaded values.

Commits are independent. Each session commits its own transaction. If task A reserves stock and task B charges a card, and B fails, A's reservation is already committed. Work that must be atomic belongs in one session, executed sequentially; fan-out is for work whose parts can succeed independently, or that has an explicit compensating step.

Reads are separate snapshots. Under READ COMMITTED, each statement sees data committed before it began. Two tasks reading related tables can observe a commit between them and produce numbers that do not reconcile. That is fine for a dashboard and wrong for an invoice.

The most common real-world boundary is between a web request and work it schedules for later. The request's session is opened by a dependency and closed when the response is sent; anything scheduled to run afterwards must open its own session. FastAPI's BackgroundTasks makes this easy to get wrong, because the task runs after the response while the dependency's session may already be closed — the full treatment is in running background tasks with a fresh AsyncSession in FastAPI.

expire_on_commit=False on the factory is close to mandatory for this style of code. With the default True, every attribute of every object expires on commit, and the first read afterwards needs a query — which under async raises instead of lazily loading. Setting it to False makes the loaded values remain readable after the session commits and closes, which is what lets a task return an object to its caller at all. The guide to expire_on_commit=False in FastAPI dependencies explains the trade-off.

Advanced Session Registries and Task-Scoped Context

Explicit async with Session() inside each task is the clearest pattern, but some codebases need a session to be available without being passed — deep call stacks, repository layers written for a request-scoped session, or libraries that call back into application code. SQLAlchemy's answer is async_scoped_session, and the important decision is what it is keyed on.

Three ways to scope a session to a task Three tiles. Explicit factory calls inside each task are the most visible and easiest to test, and cannot share by accident. async_scoped_session keyed on asyncio.current_task gives implicit per-task sessions but needs remove to be called when each task ends. A ContextVar-held session set per request is shared by every child task because tasks copy context, so it is the pattern most likely to cause sharing. explicit factory call async with Session() as s visible, testable cannot share by accident async_scoped_session scopefunc=current_task implicit per task must call remove() request ContextVar set once per request child tasks copy context so they share the session Prefer the explicit call; reach for a registry only when a framework forces the question.
import asyncio

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

engine = create_async_engine("postgresql+asyncpg://shop:secret@db/shop")
factory = async_sessionmaker(engine, expire_on_commit=False)

# One session per asyncio task.
CurrentSession = async_scoped_session(factory, scopefunc=asyncio.current_task)


async def handle_order(order_id: int) -> None:
    try:
        session = CurrentSession()
        ...  # any code in this task calling CurrentSession() gets the same session
        await session.commit()
    finally:
        # The registry holds a strong reference; without remove() it leaks.
        await CurrentSession.remove()

Keyed on asyncio.current_task, every task — including every task a request spawns — gets its own session, which is safe. The cost is the remove() call: the registry keeps each session alive until it is removed, and a forgotten remove() is a slow leak of sessions and, if they hold connections, of the pool. The existing guide on using async_scoped_session with asyncio tasks covers the cleanup patterns.

The dangerous variant is a session stored in a ContextVar set once per request. It looks equivalent and is not: asyncio.create_task() copies the current context into the new task, so every task spawned during the request reads the same ContextVar value — the same session. A registry keyed on a request ID has the identical problem. If a context variable must carry something, let it carry the factory, or a request identifier used for logging, and let each task open its own session.

A final pattern worth knowing is a per-task guard used during migration: a do_orm_execute listener that records the first task to use each session and logs when a different task appears. It finds sharing that tests miss because tests rarely produce overlapping awaits.

Hybrid Architectures & Migration Strategies

Two migrations lead teams into this topic: moving concurrent synchronous code, usually thread pools with scoped_session, onto asyncio; and moving 1.4-era async code that relied on permissive behaviour onto 2.0.

Migrating concurrent code from threads to tasks Four steps. Inventory where scoped_session is used and what scope function it relies on. Replace thread-local sessions with explicit per-task sessions from an async_sessionmaker. Replace executor fan-out with TaskGroup plus a semaphore sized from the pool. Remove blocking calls from within tasks, or move them to asyncio.to_thread, and verify with a latency-injecting test. 1 · inventory scoped_session usage thread-local registries hide which code shares a session; list every consumer 2 · make ownership explicit pass an async_sessionmaker, open sessions inside the task that uses them 3 · replace the executor ThreadPoolExecutor.map becomes TaskGroup + Semaphore(limit from the pool) 4 · prove it under latency run tests through a latency proxy so overlapping awaits actually overlap

Thread-based code tends to hide session ownership inside scoped_session, which is thread-local by default. That was safe because each thread had its own session. Porting it to asyncio by swapping in async_scoped_session with the default scope is not possible — async_scoped_session requires an explicit scopefunc precisely because there is no safe default — and choosing a request-level scope reintroduces sharing. The durable fix is to make ownership explicit while porting: pass the factory, open sessions where they are used, and replace ThreadPoolExecutor.map with a TaskGroup and a semaphore.

It is also reasonable not to port some of it. CPU-heavy or blocking work that touches the database — a report that builds a large spreadsheet, a legacy library that expects a synchronous session — can run in a thread from async code with asyncio.to_thread(), using a synchronous engine and Session. Mixing is allowed as long as each engine is used only from its own world. What is not allowed is calling synchronous database code directly on the event loop, which blocks every other task for the duration of the query.

import asyncio

from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session as SyncSession

from shop.models import Order

sync_engine = create_engine("postgresql+psycopg://shop:secret@db/shop", pool_size=2)


def build_export(month: str) -> bytes:
    with SyncSession(sync_engine) as session:
        orders = session.scalars(select(Order).where(Order.month == month)).all()
        return render_xlsx(orders)  # blocking, CPU-heavy


async def export_endpoint(month: str) -> bytes:
    return await asyncio.to_thread(build_export, month)

The same hybrid applies to Core. Concurrent bulk reads that do not need the identity map are cheaper on AsyncConnection than on AsyncSession, and the ownership rule is the same: async with engine.connect() as conn inside each task, never a connection shared across a gather().

When verifying a migration, remember that the failure mode is timing-dependent. A test suite against a local database, where every query returns in a millisecond, can pass with sessions shared everywhere. Add real network latency in tests — route the test engine through a latency proxy such as Toxiproxy, or run the suite against a database in another region — so overlapping awaits actually overlap. A time.sleep() inside an event listener does not work for this: listeners run synchronously on the event loop thread, so the sleep blocks every task and creates no overlap at all.

Observing Concurrency in Production

Concurrency bugs and concurrency costs are both invisible in ordinary request metrics. A request that holds eight connections for eighty milliseconds looks, in a latency histogram, exactly like one that holds one connection for eighty milliseconds. Three measurements make the difference visible, and together they answer the questions that matter before and after a fan-out change: how many connections does each request hold, how full is the pool, and which code is holding them.

Three numbers that make concurrency visible Three tiles. Peak connections per request, tracked with pool checkout and checkin events and a per-request tracker, shows the real fan-out factor. Pool occupancy, checked-out connections over pool_size plus max_overflow, is the leading indicator of pool timeouts. Backends per application_name in pg_stat_activity shows which service holds the database connections. peak per request checkout / checkin events a histogram per endpoint pool occupancy checkedout() ÷ capacity alert well before 1.0 backends per service pg_stat_activity grouped by application_name After a fan-out change, peak per request should rise by the fan-out factor and no further.

Peak connections per request. Count checkouts and checkins with pool events and keep the running maximum in a ContextVar set by request middleware. Because each task copies the context at creation, the counter must be a mutable object shared by reference — a one-element list or a small dataclass — rather than an integer that each task would copy.

import contextvars
from dataclasses import dataclass

from sqlalchemy import event
from sqlalchemy.ext.asyncio import AsyncEngine


@dataclass
class ConnectionUsage:
    current: int = 0
    peak: int = 0


usage: contextvars.ContextVar[ConnectionUsage | None] = contextvars.ContextVar(
    "connection_usage", default=None
)


def install_usage_tracking(engine: AsyncEngine) -> None:
    pool = engine.sync_engine.pool

    @event.listens_for(pool, "checkout")
    def _checkout(dbapi_conn, record, proxy):
        tracker = usage.get()
        if tracker is not None:
            tracker.current += 1
            tracker.peak = max(tracker.peak, tracker.current)
            record.info["tracker"] = tracker

    @event.listens_for(pool, "checkin")
    def _checkin(dbapi_conn, record):
        tracker = record.info.pop("tracker", None)
        if tracker is not None:
            tracker.current -= 1

Middleware sets usage.set(ConnectionUsage()) at the start of each request and records peak as a histogram at the end. The checkin handler reads the tracker from the connection record rather than the context, because a connection is sometimes returned from a different context than the one that checked it out.

Pool occupancy. Export pool.checkedout(), pool.overflow() and pool.size() as gauges sampled every few seconds. The ratio of checked-out connections to pool_size + max_overflow is the single best leading indicator of pool timeouts; alert well before it reaches one.

Who holds connections on the server. Set application_name per service — connect_args={"server_settings": {"application_name": "orders-api"}} — and PostgreSQL's pg_stat_activity then shows how many backends each service holds and what each one is running. During an incident, a query grouping pg_stat_activity by application_name and state tells you within seconds whether a fan-out change is responsible.

With these in place, a fan-out rollout can be judged on data: latency should drop, peak connections per request should rise by the fan-out factor and no further, and pool occupancy at peak traffic should stay comfortably below its ceiling. If occupancy climbs toward the limit, reduce the per-request cap before raising pool_size; the database's max_connections is shared by every replica of every service, and it is much harder to raise than a semaphore.

Two boundaries sit just outside the one-session-per-task rule, and both produce errors that look like session misuse. An engine's pooled connections belong to the event loop that opened them, so a process with more than one loop — a test suite, a CLI, a thread pool — needs the arrangement in using one async engine across threads and event loops. And synchronous callers that need this async stack have four ways in, compared in calling async SQLAlchemy from synchronous code — from one asyncio.run() per command to a long-lived loop in a background thread.

Production Pitfalls & Anti-Patterns

  • InterfaceError: cannot perform operation: another operation is in progress — two tasks shared one session or connection. Open a session inside each task.
  • InvalidRequestError: This session is provisioning a new connection; concurrent operations are not permitted — the same sharing, caught before the first connection was established. Same fix.
  • IllegalStateChangeError: Method 'close()' can't be called here — a request's session was closed while a task it spawned was still using it. Spawned work gets its own session and must not outlive the one it borrows from.
  • QueuePool limit of size 10 overflow 5 reached, connection timed out, timeout 30.00 after introducing fan-out — tasks per request multiplied connection demand. Cap concurrency with a semaphore sized well below pool_size.
  • ExceptionGroup: unhandled errors in a TaskGroup escaping handlers that used to catch IntegrityError — failures inside a group are wrapped. Use except* IntegrityError.
  • An asyncio.Lock around a shared session — stops the errors and silently serialises every query. Either run sequentially on purpose or use separate sessions.
Fan-out multiplies each request's pool share Bar chart against a pool of fifteen connections. A sequential request holds one. A fan-out of three holds three. A fan-out of eight holds eight. A fan-out of twenty would need more than the whole pool and would queue. sequential request 1 of 15 connections fan-out of 3 3 of 15 fan-out of 8 8 of 15 — two such requests exhaust the pool fan-out of 20 (uncapped) more than the pool: tasks queue for pool_timeout Pool of pool_size=10, max_overflow=5. Cap tasks per request well below the pool.

Fan-out also has a quieter cost that no exception reports: it raises each request's share of the pool, so a burst of fan-out requests starves ordinary ones. Watch pool.checkedout() alongside request concurrency when rolling out any parallel-query change.

Frequently Asked Questions

Is AsyncSession safe to share between asyncio tasks?

No. An AsyncSession may be used by one task at a time. Share the engine and the session factory instead, and open a session inside each task.

How many queries should I run in parallel?

No more than the pool can serve alongside everything else. A cap at or below pool_size, applied per request or per job with a semaphore, keeps a fan-out from starving other requests. Often the right number is one — a combined statement.

Does running queries concurrently make them faster?

It turns the total latency into roughly the slowest query rather than the sum, when the queries are independent. It does not make any individual query faster, and it costs one connection per concurrent query.

Can concurrent tasks write in one transaction?

No. A transaction lives on one connection and one session, which only one task may use at a time. Atomic multi-statement writes run sequentially in one session.

Should I use async_scoped_session?

Only when code needs an implicit session it cannot be passed, and only keyed on asyncio.current_task with remove() called when each task ends. Explicit factory calls are clearer and cannot share by accident.