Running queries in parallel with asyncio.TaskGroup

Open one AsyncSession inside each task created by an asyncio.TaskGroup, read results only after the async with block exits, and handle failures with except* — the group cancels the remaining queries for you when one fails. This is the structured-concurrency pattern from running concurrent queries with AsyncSession, worked through end to end.

Quick Answer

Python 3.11 added asyncio.TaskGroup, and for database fan-out it is strictly better than asyncio.gather(): when one task fails, the others are cancelled and their sessions closed before the error reaches you, so a failed request never leaves queries running behind it.

Before — gather(), with failures that leak running queries:

import asyncio

from sqlalchemy import func, select

from shop.db import Session
from shop.models import Order, Refund


async def revenue(day) -> int:
    async with Session() as session:
        return await session.scalar(
            select(func.coalesce(func.sum(Order.total_cents), 0)).where(Order.placed_on == day)
        )


async def refunds(day) -> int:
    async with Session() as session:
        return await session.scalar(
            select(func.count(Refund.id)).where(Refund.created_on == day)
        )


async def daily_summary(day):
    # If refunds() raises, revenue() keeps running and holding a connection.
    total, refunded = await asyncio.gather(revenue(day), refunds(day))
    return {"revenue_cents": total, "refunds": refunded}

After — a TaskGroup, one session per task:

import asyncio
import datetime as dt

from sqlalchemy import func, select

from shop.db import Session
from shop.models import Order, Refund


async def daily_summary(day: dt.date) -> dict[str, int]:
    async with asyncio.TaskGroup() as tg:
        revenue_task = tg.create_task(revenue(day), name="revenue")
        count_task = tg.create_task(order_count(day), name="order_count")
        refund_task = tg.create_task(refunds(day), name="refunds")
    # The block has exited: every task finished and every session is closed.
    return {
        "revenue_cents": revenue_task.result(),
        "orders": count_task.result(),
        "refunds": refund_task.result(),
    }


async def order_count(day: dt.date) -> int:
    async with Session() as session:
        return await session.scalar(
            select(func.count(Order.id)).where(Order.placed_on == day)
        )

revenue() and refunds() are unchanged from the first example. Each helper owns its session from async with to return, which is the whole rule — the group supplies the concurrency, and the helpers never see each other's connections.

Execution Context & Async Workflow Integration

Inside the group, each create_task() schedules a coroutine that will run the next time the event loop is free. The first await in each helper checks a connection out of the pool; from then on the three queries are genuinely concurrent, each waiting on its own socket. The async with asyncio.TaskGroup() block does not exit until every task has finished, succeeded or failed, so the lines after it can read results without awaiting anything.

That waiting is what makes the pattern safe with pooled connections. A connection is returned to the pool when the session that checked it out closes, and each session closes when its helper's async with Session() block exits. Because the group waits for all helpers, all three connections are back in the pool by the time the handler continues. There is no window in which a request has returned a response while its queries are still running.

The cost is connections. A request that fans out to three tasks holds three connections for the duration of the slowest query, instead of one connection for the sum of all three. At low traffic that is free latency. At high traffic it triples each request's share of the pool, and the arithmetic in sizing the asyncpg pool for high concurrency has to be redone with tasks per request as a multiplier.

Separate sessions also mean separate transactions. Under PostgreSQL's default READ COMMITTED isolation, the revenue query and the order-count query each see the data as of their own start, so an order committed between them can appear in one and not the other. For a dashboard that is acceptable. For a report that must reconcile, either combine the reads into one statement or run them sequentially in one session at REPEATABLE READ.

Resolving Warnings, Errors & Common Mistakes

Most problems with TaskGroup fan-out come from treating it like gather(): expecting a plain exception, reading results too early, or sharing a session that should have been per task.

Exception groups are the part that surprises most teams. A handler that used to catch IntegrityError from a single query needs except* once that query runs inside a group:

import asyncio

from sqlalchemy.exc import DBAPIError, IntegrityError


async def import_batch(rows_by_table):
    try:
        async with asyncio.TaskGroup() as tg:
            for table, rows in rows_by_table.items():
                tg.create_task(load_table(table, rows), name=f"load:{table}")
    except* IntegrityError as group:
        for error in group.exceptions:
            log.warning("duplicate rows skipped", extra={"detail": str(error.orig)})
        raise
    except* DBAPIError as group:
        if any(e.connection_invalidated for e in group.exceptions):
            raise RetryableImportError from group
        raise

Each except* clause receives the subset of failures matching its type, and anything not handled is re-raised in a new group. Naming tasks, as above, makes the group's traceback readable when several tables fail at once.

Advanced: Timeouts, Cancellation and Partial Results

A dashboard that waits indefinitely for its slowest aggregate is worse than one that returns without it. asyncio.timeout() bounds the whole group, and because cancellation flows through the group, a timeout cancels every still-running query rather than just abandoning them.

import asyncio
import datetime as dt
import logging

log = logging.getLogger("dashboard")


async def daily_summary_with_budget(day: dt.date, budget_s: float = 0.5) -> dict:
    results: dict[str, int | None] = {"revenue_cents": None, "orders": None, "refunds": None}

    async def collect(key: str, coro) -> None:
        results[key] = await coro

    try:
        async with asyncio.timeout(budget_s):
            async with asyncio.TaskGroup() as tg:
                tg.create_task(collect("revenue_cents", revenue(day)))
                tg.create_task(collect("orders", order_count(day)))
                tg.create_task(collect("refunds", refunds(day)))
    except TimeoutError:
        missing = [key for key, value in results.items() if value is None]
        log.warning("summary over budget", extra={"missing": missing})
    return results

When the timeout fires, each unfinished task receives CancelledError at its current await. For a task waiting on session.execute(), asyncpg sends PostgreSQL a cancel request so the statement stops on the server too, and SQLAlchemy treats the connection as unreliable — it was interrupted mid-exchange — so it is discarded rather than handed to the next request. The helper's async with Session() block then closes the session and rolls back. The finished tasks' values are already in results.

Two things undermine this. Catching CancelledError inside a helper and continuing leaves a session whose connection is gone, and the next statement fails confusingly. And wrapping a query in asyncio.shield() to "protect" it means the timeout returns while the query keeps its connection for as long as it likes — the exact leak the group was meant to prevent. Let cancellation propagate, and set a server-side statement_timeout as the backstop for queries that should never run long in any context.

A server-side limit is a one-line addition to the engine, and it catches the queries that no Python timeout is watching:

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
    "postgresql+asyncpg://shop:secret@db/shop",
    connect_args={"server_settings": {"statement_timeout": "5000"}},  # milliseconds
)

When Parallel Queries Are the Wrong Tool

Parallelism turns the sum of query times into the maximum, and that is only a win when the queries are independent, similar in duration, and cheap in connections. A surprising number of fan-outs fail at least one of those tests.

The first alternative to check is a single statement. Three aggregates over the same table are one SELECT with three expressions; three aggregates over different tables are scalar subqueries in one SELECT. PostgreSQL runs them in one round trip on one connection, often faster than three parallel queries because there is no connection overhead at all:

import datetime as dt

from sqlalchemy import func, select

from shop.db import Session
from shop.models import Order, Refund


async def daily_summary_single(day: dt.date) -> dict[str, int]:
    revenue = (
        select(func.coalesce(func.sum(Order.total_cents), 0))
        .where(Order.placed_on == day)
        .scalar_subquery()
    )
    orders = select(func.count(Order.id)).where(Order.placed_on == day).scalar_subquery()
    refunds = select(func.count(Refund.id)).where(Refund.created_on == day).scalar_subquery()
    async with Session() as session:
        row = (await session.execute(select(revenue, orders, refunds))).one()
    return {"revenue_cents": row[0], "orders": row[1], "refunds": row[2]}

The second is caching. An aggregate that changes hourly does not need to be computed on every request, in parallel or otherwise.

Fan-out earns its place when the work genuinely cannot be combined: reads from two different databases, a database read interleaved with an HTTP call, or loading several large, independent result sets for an export. For those, the TaskGroup pattern above is the right shape, and bounding it with a semaphore — as in the guide to fixing "another operation is in progress" errors — keeps it from starving the rest of the service.

Frequently Asked Questions

Related

• Running Concurrent Queries with AsyncSession — The parent guide: task boundaries, session factories and fan-out design.
• Fixing "another operation is in progress" errors — The error a shared session produces, and bounding fan-out with a semaphore.
• Setting transaction isolation level per session — When parallel reads must agree with each other.
• Setting up asyncpg pool size for high concurrency — Redoing the pool arithmetic once requests fan out.