Structured concurrency

Structured concurrency keeps child tasks inside a lexical scope: the parent does not finish until its children finish, and failures trigger predictable cleanup. Python 3.11 introduced asyncio.TaskGroup; it should be the default for a related set of child tasks.

TaskGroup

import asyncio

async def fetch(item_id: int) -> str:
    await asyncio.sleep(0.05)
    return f"item-{item_id}"

async def fetch_all(ids: list[int]) -> list[str]:
    tasks: list[asyncio.Task[str]] = []
    async with asyncio.TaskGroup() as group:
        for item_id in ids:
            tasks.append(group.create_task(fetch(item_id), name=f"fetch-{item_id}"))
    return [task.result() for task in tasks]

On the first non-cancellation failure, a task group cancels unfinished siblings and waits for them. It then raises an ExceptionGroup. Results are available after successful scope exit.

try:
    results = await fetch_all([1, 2, 3])
except* TimeoutError as errors:
    for error in errors.exceptions:
        print(error)

Use except* to handle matching members of an exception group. Do not flatten failures into an unstructured string when callers need to distinguish causes.

gather

asyncio.gather() preserves input order and is useful when you want one aggregate awaitable:

results = await asyncio.gather(*(fetch(item_id) for item_id in [3, 1, 2]))

Its failure semantics differ from TaskGroup. With default settings, the first propagated exception does not automatically cancel other submitted awaitables. return_exceptions=True places exceptions in the result list, which can easily hide failures:

outcomes = await asyncio.gather(*operations, return_exceptions=True)
for outcome in outcomes:
    if isinstance(outcome, BaseException):
        handle_failure(outcome)

Choose TaskGroup for sibling tasks that form one operation. Choose gather when its aggregate result and explicit failure semantics are exactly what you need.

Cancellation is control flow

Task.cancel() requests cancellation. CancelledError is injected at a suspension point and inherits directly from BaseException, not Exception.

import asyncio

async def worker() -> None:
    resource = await acquire_resource()
    try:
        await use_resource(resource)
    finally:
        await release_resource(resource)

Use try/finally or context managers for cleanup. If you catch CancelledError, normally clean up and re-raise:

async def worker() -> None:
    try:
        await do_work()
    except asyncio.CancelledError:
        await record_interruption()
        raise

Swallowing cancellation can break TaskGroup, timeout scopes, and shutdown. If code intentionally suppresses it, it may also need Task.uncancel(), but that is an advanced and uncommon design.

Timeout scopes

import asyncio

async def bounded_operation() -> str:
    try:
        async with asyncio.timeout(2.0):
            return await slow_operation()
    except TimeoutError:
        return "timed out"

asyncio.timeout() cancels the current task inside the context and converts that cancellation to TimeoutError outside the context. Catch TimeoutError after the async with, not inside it.

asyncio.wait_for(awaitable, timeout) remains useful for one awaitable. It cancels the operation on expiry and waits for cancellation cleanup, so wall-clock duration can exceed the nominal timeout.

Use asyncio.timeout_at(loop.time() + seconds) to share an absolute monotonic deadline across nested operations, preventing each layer from receiving a fresh full timeout.

Shielding

asyncio.shield() prevents an inner awaitable from being cancelled when its waiter is cancelled, but the waiter still receives CancelledError. Keep a task reference. Shield only short, well-owned operations such as a critical protocol acknowledgement; widespread shielding defeats structured cancellation.

Graceful shutdown

The owner should stop accepting work, signal workers, let bounded cleanup run, and then exit. Never rely on garbage collection to close tasks or transports. Every long-lived task needs:

• an owner;
• a stop condition;
• a bounded shutdown path;
• a policy for exceptions.

Practice

Fetch ten independent records under one five-second deadline. Limit concurrency, cancel siblings if any record is invalid, and preserve results in input order. Add one worker that delays cleanup to observe why timeout completion may exceed the nominal deadline.