Run Related Tasks Together with TaskGroup

You fetch three pages at once with asyncio.gather, and one of them fails. Here fetch sleeps to stand in for a download, and the page "b" raises:

import asyncio


async def fetch(name: str, delay: float) -> str:
    try:
        await asyncio.sleep(delay)
    except asyncio.CancelledError:
        print(f"{name} cancelled")
        raise
    if name == "b":
        raise ValueError(f"{name} failed")
    print(f"{name} finished")
    return name


async def main() -> None:
    try:
        await asyncio.gather(fetch("a", 0.2), fetch("b", 0.1), fetch("c", 0.2))
    except ValueError as exc:
        print("caught:", exc)
    await asyncio.sleep(0.3)


asyncio.run(main())

caught: b failed
a finished
c finished

gather raised the error as soon as "b" failed. Your except block handled it, and "a" and "c" kept running anyway. In real code, those are requests still holding connections, or writes that finish after you reported a failure. If two pages fail, you see only the first error.

A TaskGroup stops the other tasks

asyncio.TaskGroup ties a set of tasks to one async with block. Start each task with tg.create_task:

async def main() -> None:
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fetch("a", 0.2))
            tg.create_task(fetch("b", 0.1))
            tg.create_task(fetch("c", 0.2))
    except* ValueError as group:
        print("failed:", group.exceptions)

a cancelled
c cancelled
failed: (ValueError('b failed'),)

The two cancelled lines can come in either order. Three rules explain the output:

The async with block does not exit until every task in the group has finished.
When a task fails, the group cancels the others. Cancelling a task asks it to stop. It gets a CancelledError at its current await, which is why fetch printed cancelled. A later article covers cancellation in depth.
Once every task has stopped, the group raises an ExceptionGroup, an exception that holds a list of exceptions. It holds every failure, not only the first. except* ValueError catches the ValueErrors inside it.

When nothing fails, read each result with task.result() after the block:

async with asyncio.TaskGroup() as tg:
    tasks = {name: tg.create_task(fetch(name, 0.1)) for name in ["a", "c"]}
pages = {name: task.result() for name, task in tasks.items()}

Choose a failure policy

Before you write the code, decide what one failure means for the whole call. Two policies cover most code.

All or nothing. If any fetch fails, the whole call fails and nothing is left running. Use a TaskGroup as above.

Partial results. Every fetch runs to the end, and you return the successes and the failures separately. Catch the error inside each task, so it never reaches the group:

async def fetch_many(names: list[str]) -> tuple[dict[str, str], dict[str, Exception]]:
    pages: dict[str, str] = {}
    errors: dict[str, Exception] = {}

    async def one(name: str) -> None:
        try:
            pages[name] = await fetch(name, 0.1)
        except Exception as exc:
            errors[name] = exc

    async with asyncio.TaskGroup() as tg:
        for name in names:
            tg.create_task(one(name))
    return pages, errors

The group still owns the tasks. If the caller of fetch_many is cancelled, the group cancels every task and waits for them before it lets the cancellation through.

gather(..., return_exceptions=True) also gives partial results. It returns a list with a value or an exception object in each position. Then you can't tell a raised exception from one that a fetch returned as its value.

Common mistakes

Calling task.result() inside the async with block. The task may not be done yet, and you get InvalidStateError. Read results after the block.
Starting a task with asyncio.create_task inside the block. That task is not in the group, so the group neither waits for it nor cancels it. Use tg.create_task.
Catching a group failure with except ValueError. A single failure still arrives wrapped in an ExceptionGroup. Use except* ValueError.

Check your understanding Three fetches run in a TaskGroup and the second one fails. What happens to the other two, and what does the caller see? Show answer

The group cancels the other two and waits until they have stopped. Then it raises an ExceptionGroup that holds the second fetch's exception. The caller catches it with except*. When the call returns, nothing from it is still running.

Going further

A TaskGroup does not wrap KeyboardInterrupt or SystemExit in an ExceptionGroup. It cancels its tasks and re-raises the exception as it is.

A task that raises CancelledError itself doesn't count as a failure, so the group doesn't cancel the other tasks. Reading its result with task.result() raises CancelledError.

mypy keeps the result type through gather. With fetch(key: str) -> bytes, await asyncio.gather(*(fetch(key) for key in keys)) has type list[bytes].

See the TaskGroup documentation.