TaskGroup and Structured Concurrency

asyncio.TaskGroup (Python 3.11+) is the modern replacement for asyncio.gather when you want structured concurrency — guaranteed cleanup, propagated exceptions, no leaked tasks.

The gather problem

async def main():
    results = await asyncio.gather(
        fetch_a(),
        fetch_b(),
        fetch_c(),
    )

What if fetch_b() raises mid-flight? gather by default:

• Cancels nothing automatically.
• Returns the exception from fetch_b.
• Leaves fetch_a and fetch_c running invisibly until they finish or the loop dies.

Adding return_exceptions=True swallows them silently. Either way, you have to write the cancellation + cleanup logic yourself.

TaskGroup — the structured way

async def main():
    async with asyncio.TaskGroup() as tg:
        a = tg.create_task(fetch_a())
        b = tg.create_task(fetch_b())
        c = tg.create_task(fetch_c())
    # All three are done by the time we exit the `async with`.
    print(a.result(), b.result(), c.result())

Guarantees:

• The async with block doesn’t exit until every task completes (success, failure, or cancellation).
• If any task raises, all sibling tasks are cancelled, and the exceptions are collected into an ExceptionGroup.
• No leaked tasks. Ever.

try:
    async with asyncio.TaskGroup() as tg:
        tg.create_task(fetch_a())
        tg.create_task(failing())
        tg.create_task(slow())   # will be cancelled when failing() raises
except* ValueError as eg:
    # except* matches all ValueErrors inside the ExceptionGroup
    for e in eg.exceptions:
        log.error("ValueError in subtask", exc_info=e)
except* TimeoutError as eg:
    ...

except* (PEP 654) splits the ExceptionGroup by type. Old try/except still catches the whole ExceptionGroup but doesn’t see individual types.

Why “structured” matters

Structured concurrency = the lifetime of every concurrent task is bounded by a syntactic block. You always know exactly where a task can be running. You can’t “fire and forget” inside a TaskGroup — even tasks created inside it block the exit.

Practical effects:

• No leaked tasks. Asyncio’s “Task was destroyed but it is pending” warning vanishes.
• Cleanup is automatic. If you raise, sibling work stops.
• Cancellation is correct. TaskGroup propagates cancellation downward.

Same idea in Trio (nursery) and anyio (create_task_group). Trio invented the term; Python 3.11 imported it as TaskGroup.

Cancellation in a TaskGroup

async def main():
    async with asyncio.TaskGroup() as tg:
        a = tg.create_task(asyncio.sleep(60), name="long-job")
        b = tg.create_task(fetch_critical_data())
    # If main() is cancelled here, both a and b are cancelled and awaited.

A CancelledError raised from outside cancels every task in the group. They get a chance to clean up (finally blocks run), then the TaskGroup re-raises a single CancelledError.

When to still use gather

For 3.11+, default to TaskGroup. Use gather for one-liners where the all-or-nothing semantic is what you want.

Picking the right primitive

The whole asyncio task API in one table. Most “which one do I use” questions resolve here.

Two rules that catch most mistakes:

1. Never block the loop. time.sleep, requests, sync DB drivers, and heavy CPU loops stall every coroutine in the process. Use asyncio.sleep, an async client, or push it to a thread/process.
2. Every task needs an owner. Something must await it and observe its exception. A task nobody awaits is a task whose failure you’ll never see.

Common patterns

Fan-out with rate-limited spawn

async def process_all(items):
    sem = asyncio.Semaphore(20)
    async def one(item):
        async with sem:
            return await process(item)

    async with asyncio.TaskGroup() as tg:
        for item in items:
            tg.create_task(one(item))

Producer-consumer

async def main():
    queue = asyncio.Queue(maxsize=100)
    async with asyncio.TaskGroup() as tg:
        tg.create_task(producer(queue))
        for _ in range(10):
            tg.create_task(consumer(queue))
        # Send sentinels when producer is done, OR cancel consumers explicitly.

Timeouts on the whole group

async with asyncio.timeout(30):
    async with asyncio.TaskGroup() as tg:
        tg.create_task(slow_fetch())
        tg.create_task(other_fetch())

asyncio.timeout (3.11+) cancels everything inside on timeout; TaskGroup propagates that.

Gotchas

• Tasks created in a TaskGroup must use tg.create_task, not asyncio.create_task — otherwise they’re outside the group’s lifetime and leak as before.
• Don’t await raw tasks inside the group body when you don’t need to — let the group’s async with exit do it. Awaiting inside ties you to that task’s success.
• except* requires Python 3.11+. On older versions you can still use TaskGroup-style patterns via anyio or write it manually.
• return_exceptions=True equivalent doesn’t exist. TaskGroup is all-or-nothing — if you want to collect partial results with errors, gather is still your friend.

anyio — works on Trio and asyncio

import anyio

async def main():
    async with anyio.create_task_group() as tg:
        tg.start_soon(fetch_a)
        tg.start_soon(fetch_b)

Same structured concurrency, works on either backend. Useful if you want code to run unchanged on Trio.

Interview angle

• “What does TaskGroup give you that gather doesn’t?” — guaranteed structured concurrency: tasks can’t outlive the block; one failure cancels siblings; exceptions are collected as ExceptionGroup. With gather you have to write the cancellation/cleanup logic yourself.
• “What’s structured concurrency?” — every concurrent task’s lifetime is bounded by a syntactic block; you always know where work is running and that it will be cleaned up. Trio invented the term; Python 3.11 ported it as TaskGroup.
• “How do you catch one specific exception from a TaskGroup?” — try/except* ValueError as eg: (PEP 654). The * matches inside ExceptionGroup; iterate eg.exceptions to inspect individuals.
• “When would you still use gather?” — older Python (≤3.10), one-line fan-out where all-or-nothing failure is fine, or return_exceptions=True for “best effort, give me what worked”.
• “What’s asyncio.timeout and how does it compose with TaskGroup?” — context manager that cancels everything inside on timeout. Wrapping a TaskGroup in asyncio.timeout(30) gives you “cancel all subtasks if not done in 30s”.