Skip to content

Why One SQLAlchemy AsyncSession Cannot Run Queries Concurrently

asyncio.gather over several queries on the same AsyncSession looks like free parallelism, and it is the most common SQLAlchemy asyncio mistake. A session wraps one database connection and one unit of work — identity map, pending objects, transaction state — none of which is safe to drive from several tasks at once. Tested with SQLAlchemy 2.1 and asyncpg against PostgreSQL 17, the outcome depended on timing, which is what makes the bug hard to find: on a fresh session, three concurrent execute calls raised InvalidRequestError: This session is provisioning a new connection; concurrent operations are not permitted. On a session that already held a connection, ten concurrent 50 ms queries did not fail — they ran one after another, in 0.51 s, the same as a loop. Ten concurrent add + flush calls failed 9 times out of 10 with Session is already flushing. Ten separate sessions ran the same ten queries in 0.16 s. This guide explains why and shows the correct shapes.

Prerequisites

1. See the failure modes

The same mistake produces three different behaviours depending on the session's state:

async with Session() as session:
    # fresh session: no connection yet
    await asyncio.gather(*(session.execute(stmt) for _ in range(3)))
    # InvalidRequestError: This session is provisioning a new connection;
    # concurrent operations are not permitted


async with Session() as session:
    await session.execute(text("select 1"))            # connection now checked out
    async with asyncio.TaskGroup() as tg:
        for _ in range(10):
            tg.create_task(session.execute(text("select pg_sleep(0.05)")))
    # no error, but measured 0.507 s: the ten queries ran one at a time

The first fails loudly because two tasks try to check out the session's connection at once. The second "works" because the asyncpg adapter serializes statements on the connection with a lock — so the code passes tests and silently delivers no concurrency. Neither gives what the author wanted.

Verify: add timing around a gather on one session; if it equals the sum of the query times, the queries were serialized.

Ten 50 ms queries, by session layout 3 horizontal bars comparing one session, loop with the others. Ten 50 ms queries, by session layout one session, loop 0.511 s one session, 10 tasks 0.507 s, serialized 10 sessions, 10 tasks 0.158 s SQLAlchemy 2.1.2, asyncpg 0.31, PostgreSQL 17; select pg_sleep(0.05). Tasks sharing a session wait for each other on its single connection.

2. Understand why it cannot work

A session is a unit of work over a single connection and transaction. Three things are inherently sequential:

session = Session()
# 1. one connection: PostgreSQL runs one statement at a time per connection
# 2. one transaction: all statements belong to it; one task's error rolls back everyone's work
# 3. one identity map and flush: pending objects are written in one flush at a time

async def add_and_flush(i):
    session.add(Item(id=i))
    await session.flush()         # measured: 9 of 10 concurrent calls raised
                                  # InvalidRequestError: Session is already flushing

The flush failure is the clearest illustration: a flush writes all pending objects in the session, so a second task calling flush() mid-flush would see a half-written state. SQLAlchemy refuses it. Even where no error is raised, the transaction is shared — if one task's statement fails, the transaction is aborted for all of them, and the others' writes are rolled back with it. The SQLAlchemy documentation states the rule plainly: an AsyncSession is not safe for use in concurrent tasks.

Verify: run concurrent flushes on one session in a test; it raises, confirming that the code path must be changed rather than retried.

3. Use one session per concurrent task

When work genuinely needs to run in parallel, give each task its own session from the sessionmaker. Each gets its own connection from the engine's pool:

from sqlalchemy.ext.asyncio import async_sessionmaker

Session = async_sessionmaker(engine, expire_on_commit=False)


async def load_dashboard(user_id: int) -> dict:
    async def run(stmt):
        async with Session() as s:                # its own connection and transaction
            return (await s.execute(stmt)).scalars().all()

    async with asyncio.TaskGroup() as tg:
        orders = tg.create_task(run(select(Order).where(Order.user_id == user_id)))
        invoices = tg.create_task(run(select(Invoice).where(Invoice.user_id == user_id)))
        alerts = tg.create_task(run(select(Alert).where(Alert.user_id == user_id)))
    return {"orders": orders.result(), "invoices": invoices.result(), "alerts": alerts.result()}

Measured: ten queries in ten sessions took 0.158 s against 0.51 s serialized. The price is one pooled connection per concurrent task, so a request that fans out to three queries needs three connections — size the pool for it, as in sizing async connection pools for throughput. The three sessions are separate transactions, so they do not see each other's uncommitted writes; use this for independent reads, not for writes that must commit together.

Verify: the fan-out completes in about the time of the slowest query, and the pool's checked-out count rises by the number of tasks.

Shared session versus session per task A sequence of 6 messages between 4 participants. Shared session versus session per task task A task B shared session engine pool execute: holds the connection execute: waits for A B runs after A finishes own session: connection 1 own session: connection 2 both run at once Parallel statements need parallel connections, which means separate sessions.

4. Keep writes in one session, sequentially

When the statements must commit atomically — create an order, its lines and an audit row — they belong in one session and one transaction, run sequentially. That is not a performance compromise: the database runs one transaction's statements in order regardless.

async def place_order(session, data) -> Order:
    order = Order(customer_id=data.customer_id)
    session.add(order)
    await session.flush()                          # get order.id
    session.add_all([OrderLine(order_id=order.id, **line) for line in data.lines])
    session.add(Audit(action="order.created", ref=order.id))
    await session.commit()                         # all or nothing
    return order

If the write path is slow, the fix is fewer round trips — add_all and a single flush, insert().values([...]) for bulk rows, or a COPY for large loads — not concurrency within the session. Fan out only the independent reads that come before or after the transaction.

Verify: a failure in the audit insert rolls back the order and its lines together.

5. Find shared-session concurrency in existing code

The bug hides behind helpers: a function that takes a session argument gets called from several tasks with the same session. Look for it and guard against it:

class GuardedSession:
    """Development-only wrapper: raises if two tasks use the same session at once."""

    def __init__(self, session) -> None:
        self._s, self._owner = session, None

    async def execute(self, *a, **kw):
        task = asyncio.current_task()
        if self._owner not in (None, task):
            raise RuntimeError("AsyncSession used from two tasks concurrently")
        self._owner = task
        try:
            return await self._s.execute(*a, **kw)
        finally:
            self._owner = None

In code review, flag asyncio.gather(...) or TaskGroup blocks whose coroutines receive the same session. In request handlers, a session obtained from a dependency is per request, so it must never be handed to background tasks that outlive or run alongside the handler — those create their own, as in scoping database sessions with contextvars.

Verify: run the test suite with the guard enabled; any concurrent use raises with a stack trace pointing at the call site.

How should these queries run? A decision on What are the statements with 3 outcomes. How should these queries run? What are the statements? independent reads session per task parallel, separate transactions writes that commit together one session sequential, one transaction background work from a request new session never the request's Concurrency belongs between sessions, never inside one.

Verification

Sessions are used correctly when:

  • No session is shared between concurrently running tasks.
  • Parallel reads use one session per task, with the pool sized for the fan-out.
  • Atomic writes stay in one session, executed sequentially.
  • A development guard or test detects concurrent use.

Diagnostic Hook: search logs for concurrent operations are not permitted and Session is already flushing — both are certain signs of a shared session. For the silent case, compare a fan-out endpoint's latency with the sum of its queries' durations from database logs; if they match, the queries were serialized on one session.

Pitfalls & edge cases

  • Testing with a warm session. The shared-session bug can pass tests and only fail on first use.
  • Fanning out writes into separate sessions. They become separate transactions that can partially commit.
  • Passing the request's session to background tasks. It may be closed or in use.
  • Forgetting the pool. Each parallel session needs its own connection.

Frequently Asked Questions

Can I use asyncio.gather with one SQLAlchemy AsyncSession?

No. On a fresh session it raises InvalidRequestError about concurrent operations; on a session that already has a connection the queries run one at a time; concurrent flushes fail with "Session is already flushing". Use one session per task.

How do I run SQLAlchemy queries in parallel with asyncio?

Create a separate AsyncSession for each concurrent task from the async_sessionmaker. In testing, ten 50 ms queries took 0.16 s in ten sessions versus 0.51 s on one.

Why does gather on one session not give an error sometimes?

Once the session holds a connection, the asyncpg adapter serializes statements on it, so they run sequentially without error. It hides the bug rather than making it safe.

Should related writes go into separate sessions to run them in parallel?

No. Separate sessions are separate transactions that can partially commit. Keep writes that must succeed together in one session, executed in order.