Fixing "another operation is in progress" errors with asyncpg

Give every concurrent task its own AsyncSession from an async_sessionmaker — asyncio.gather() over one shared session sends two statements down one asyncpg connection, and asyncpg refuses the second with InterfaceError: cannot perform operation: another operation is in progress. This guide is the error-by-error companion to running concurrent queries with AsyncSession.

Quick Answer

The error is not a driver bug and not a pool problem. It means two coroutines were using the same connection at the same moment, and the only way that happens under SQLAlchemy is two coroutines sharing one AsyncSession (or one AsyncConnection).

Before — one session fanned out across tasks:

import asyncio

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from shop.models import Customer, Order


async def load_dashboard(session: AsyncSession, customer_id: int):
    # Both coroutines share `session`, so both share its single connection.
    customer, orders = await asyncio.gather(
        session.get(Customer, customer_id),
        session.scalars(select(Order).where(Order.customer_id == customer_id)),
    )
    return customer, orders.all()
# sqlalchemy.exc.InterfaceError: (sqlalchemy.dialects.postgresql.asyncpg.InterfaceError)
# <class 'asyncpg.exceptions._base.InterfaceError'>:
# cannot perform operation: another operation is in progress

After — one session per task, from the factory:

import asyncio

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

from shop.models import Customer, Order

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


async def get_customer(customer_id: int) -> Customer | None:
    async with Session() as session:
        return await session.get(Customer, customer_id)


async def get_orders(customer_id: int) -> list[Order]:
    async with Session() as session:
        result = await session.scalars(select(Order).where(Order.customer_id == customer_id))
        return list(result)


async def load_dashboard(customer_id: int):
    return await asyncio.gather(get_customer(customer_id), get_orders(customer_id))

The objects each task returns are detached once its session closes, which is why the factory sets expire_on_commit=False and the queries load everything the caller will read. If the caller needs a relationship, load it inside the task with selectinload() rather than touching it afterwards.

Execution Context & Async Workflow Integration

A PostgreSQL connection is a single conversation. The client sends a query, the server streams back rows, and only when the server says it is ready does the next query begin. asyncpg enforces that order explicitly: each connection records whether an operation is outstanding, and a second call before the first completes raises InterfaceError without sending anything.

Under synchronous code this could never happen by accident, because a thread blocked on a socket read cannot issue another query. Under asyncio it happens easily. await session.execute() suspends the task while it waits for rows; the event loop runs the next ready task; if that task holds a reference to the same session, it reaches the same connection and trips the guard.

AsyncSession does not add a lock to paper over this, deliberately. A session is a unit of work — an identity map, a pending-change list and a transaction — and two tasks interleaving writes into one unit of work would produce a transaction whose contents depend on scheduling. The documentation states the rule plainly: a single AsyncSession is not safe for use in concurrent tasks. The AsyncSession lifecycle is designed around one session per logical operation, and concurrency is achieved by running several of those operations, not by sharing one.

The same reasoning applies one level down. An AsyncConnection obtained from engine.connect() is also a single conversation, so asyncio.gather() over conn.execute() calls fails the same way. The fix there is async with engine.connect() as conn inside each task.

Where the tasks come from does not matter. asyncio.gather(), asyncio.TaskGroup, asyncio.create_task(), an anyio task group, a background task launched from a request handler that closes over the request's session — all of them create the same overlap. The last is the most common in web applications, and it is covered in its own guide on background tasks in FastAPI.

Resolving Warnings, Errors & Common Mistakes

Four different exceptions describe this one mistake, depending on what the session was doing when the second task arrived. Recognising them as the same bug saves a lot of time spent tuning pools that are not the problem.

The last row matters. Moving from one shared session to one session per task converts a correctness bug into a capacity question: a gather() over two hundred tasks now wants two hundred connections. The pool will queue them — pool_timeout defaults to thirty seconds — but a request that holds ten connections while other requests wait is a pool exhaustion incident waiting to happen.

A second mistake is "fixing" the error with an asyncio.Lock around the shared session. It does stop the exception, and it also serialises every query, so the gather() now runs exactly as slowly as sequential code while looking concurrent. If the operations must share a transaction, run them sequentially on purpose; if they do not, give them separate sessions.

Advanced: Bounding the Fan-out to the Pool

Once each task has its own session, the useful question is how many to run at once. The answer comes from the pool, not from the number of items: concurrency above pool_size + max_overflow does not run faster, it only waits in the pool queue where the wait is harder to see.

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

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

from shop.models import Product

T = TypeVar("T")
R = TypeVar("R")

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


async def bounded_map(
    func: Callable[[T], Awaitable[R]], items: Iterable[T], limit: int
) -> list[R]:
    semaphore = asyncio.Semaphore(limit)

    async def run(item: T) -> R:
        async with semaphore:
            return await func(item)

    async with asyncio.TaskGroup() as group:
        tasks = [group.create_task(run(item)) for item in items]
    return [task.result() for task in tasks]


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


async def price_many(skus: list[str]) -> dict[str, int | None]:
    # Leave headroom: other requests share this pool.
    return dict(await bounded_map(price_of, skus, limit=4))

Two details in that helper are deliberate. The semaphore is acquired before the session is opened, so a waiting task holds nothing — no connection, no transaction. And TaskGroup rather than gather() means the first failure cancels the siblings instead of leaving them running against the database after the caller has already raised.

Before reaching for fan-out at all, check whether one statement would do. Two independent lookups by key are often a single query with a join or an IN list, which uses one connection and one round trip. Concurrency across sessions is for work that genuinely cannot be combined — separate aggregates, separate databases, or calls interleaved with non-database I/O.

Finding the Shared Session in an Existing Codebase

The error message names the connection, not the code that shared it, so in a large service the first job is finding where one session reaches two tasks. Three places account for nearly every case, and each can be checked mechanically.

Session parameters passed into spawned work. Search for create_task(, gather(, TaskGroup and BackgroundTasks.add_task( and read the arguments. Any call that passes session, db or a repository object that wraps one is a candidate. Repository classes hide the problem well: OrderRepository(session) looks like a plain object, but two tasks calling its methods are two tasks on one connection.

Module-level or class-level sessions. A session created at import time, or stored on a long-lived service object, is shared by every request that touches it. grep -rn "= AsyncSession(" --include=*.py and grep -rn "Session()" --include=*.py outside a with block find most of these.

Scoped sessions keyed on the wrong thing. An async_scoped_session keyed on current_task gives each task its own session, which is correct. One keyed on a request-id ContextVar hands the same session to every task spawned during the request, because child tasks copy the parent's context. That is exactly the sharing this guide is about, arrived at by configuration rather than by code.

A cheap runtime check confirms the diagnosis before any refactor. Record the task that first used each session and complain when another task shows up:

import asyncio
import logging

from sqlalchemy import event
from sqlalchemy.orm import Session

log = logging.getLogger("db.task_guard")


@event.listens_for(Session, "do_orm_execute")
def _guard_task(orm_execute_state) -> None:
    session = orm_execute_state.session
    task = asyncio.current_task()
    owner = session.info.setdefault("owner_task", task)
    if owner is not task:
        log.error(
            "session used by more than one task",
            extra={"owner": owner.get_name() if owner else None,
                   "intruder": task.get_name() if task else None},
            stack_info=True,
        )

The listener attaches to the synchronous Session class, which is what every AsyncSession wraps, and the logged stack points at the intruding call site. Run it in staging for a day, fix what it reports, and remove it — it costs a dictionary lookup per statement, which is small but not free.

Frequently Asked Questions

Related

• Running Concurrent Queries with AsyncSession — The parent guide: task boundaries, session factories and fan-out design.
• Running queries in parallel with asyncio.TaskGroup — Structured concurrency, cancellation and error handling for parallel reads.
• Debugging QueuePool limit reached timeouts — What happens when the fan-out outgrows the pool.
• Using async_scoped_session with asyncio tasks — Task-keyed session registries, and when they help.