Using Python TaskGroup for Structured Concurrency

Python's asyncio.TaskGroup (introduced in Python 3.11) brings structured concurrency to asyncio. It manages a collection of tasks as a single unit, ensuring that all tasks complete or are cancelled together, and that exceptions are propagated predictably. This article explains how to use python taskgroup effectively, how it differs from asyncio.gather, and where it fits in production code.

Basic Usage of asyncio.TaskGroup

A TaskGroup is created using an async with block. Tasks are added by calling create_task() on the group. The block exits only after all tasks have finished. Here is a minimal example:

import asyncio

async def fetch_data(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"{name} done"

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(fetch_data("first", 1.0))
        task2 = tg.create_task(fetch_data("second", 2.0))
    print(task1.result())
    print(task2.result())

asyncio.run(main())

Inside the async with block, you can create as many tasks as needed. When the block exits normally (i.e., no exception is raised), all tasks have completed successfully. The result() method on each task returns its return value, just like with a regular asyncio.Task.

The key difference from manually managing a list of tasks is that the TaskGroup waits for every task to finish before continuing. If any task raises an exception, the group handles it according to structured concurrency rules.

Error Handling and Automatic Cancellation

One of the main benefits of TaskGroup is its exception behavior. If any task inside the group raises an exception, the group cancels all remaining tasks that have not yet completed. This prevents orphaned tasks from continuing after a failure. The first exception is then propagated out of the async with block.

Consider this example:

import asyncio

async def fail():
    raise ValueError("boom")

async def slow_task():
    await asyncio.sleep(10)
    return "done"

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fail())
            tg.create_task(slow_task())
    except* ValueError as eg:
        print(f"Caught: {eg.exceptions}")

asyncio.run(main())

When fail() raises, the TaskGroup immediately cancels slow_task() and then raises an ExceptionGroup containing the original exception. The except* syntax (also new in Python 3.11) allows you to catch specific exception types from the group. Without except*, the ExceptionGroup would propagate unchanged, which can be cumbersome if you only care about one type.

This behavior is fundamentally different from asyncio.gather, which does not cancel other tasks when one fails. With gather, you must manually cancel the remaining tasks if you want that behavior. TaskGroup makes failure atomic: either all tasks succeed, or the group aborts with an exception.

Cancellation of the TaskGroup Itself

A TaskGroup can also be cancelled from outside. If the coroutine that owns the async with block is cancelled, the group cancels all its child tasks and waits for them to finish cancellation before propagating the CancelledError. This ensures that cleanup happens in a controlled manner.

import asyncio

async def worker():
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("worker cancelled")
        raise

async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(worker())
        await asyncio.sleep(0.1)
        raise asyncio.CancelledError()

asyncio.run(main())

In this example, the main() coroutine raises CancelledError after starting a task. The TaskGroup catches that cancellation, cancels worker(), and waits for it to finish. The CancelledError is then re-raised. This makes it safe to use TaskGroup inside code that may be cancelled, because you know all child tasks will be stopped before the group exits.

TaskGroup vs asyncio.gather

Both TaskGroup and asyncio.gather run multiple coroutines concurrently, but they differ in error handling and return semantics.

gather is useful when you want to collect results from several coroutines and you don't mind that a failure leaves other tasks running. For example, if you are fetching independent resources and want to handle partial success, gather may be simpler. But if you need to guarantee that a failure stops all work, TaskGroup is the safer choice.

Another distinction: gather returns a list of results, while TaskGroup gives you the Task objects directly. This is convenient when you need to inspect individual task states or call methods like result() after the group completes.

Nested Task Groups

TaskGroup can be nested. When an inner group is created inside an outer group, the inner group's tasks are also managed by the outer group. This is useful for hierarchical cancellation and error propagation.

import asyncio

async def inner_work():
    await asyncio.sleep(1)
    return "inner"

async def outer_work():
    async with asyncio.TaskGroup() as inner_tg:
        inner_tg.create_task(inner_work())

async def main():
    async with asyncio.TaskGroup() as outer_tg:
        outer_tg.create_task(outer_work())

asyncio.run(main())

If a task in the inner group fails, the inner group cancels its remaining tasks and raises an ExceptionGroup. That exception is then caught by the outer group, which treats it as a task failure and cancals its other tasks. This ensures that errors propagate correctly through multiple levels of nesting.

Be careful with nesting when you need to handle exceptions at different levels. Use except* at the appropriate scope to catch specific exception types without swallowing unrelated errors.

Compatibility and Practical Considerations

TaskGroup requires Python 3.11 or later. If you are on an older version, you can use the trio library's CancelScope or asyncio.gather with manual cancellation, but the ergonomics are not as clean. When upgrading, note that ExceptionGroup is also new in 3.11, so code that catches exceptions from TaskGroup must use except* or handle ExceptionGroup directly.

One practical limitation is that you cannot add tasks to a TaskGroup after the async with block has exited. The group is closed once the block ends. This means you must decide the set of tasks upfront, or use a loop that creates tasks inside the block. For dynamic task creation, consider using a queue and worker tasks inside the group.

Another consideration is that TaskGroup does not provide a way to retrieve results in order. You must keep references to the Task objects and call result() on each. If you need results in a specific order, store the tasks in a list and iterate after the group completes.

Finally, remember that TaskGroup is designed for structured concurrency: it enforces that all child tasks finish before the group exits. This is a good default for most applications, but if you intentionally want to leave tasks running after a failure, gather or manual task management is more appropriate.

Using TaskGroup with Timeouts and External Signals

A common pattern is to combine TaskGroup with asyncio.timeout to bound the total execution time of a set of tasks. Because TaskGroup cancels all tasks when an exception occurs, you can wrap the group in a timeout to ensure that no task runs longer than allowed.

import asyncio

async def long_task():
    await asyncio.sleep(30)

async def main():
    try:
        async with asyncio.timeout(2):
            async with asyncio.TaskGroup() as tg:
                tg.create_task(long_task())
    except TimeoutError:
        print("Timed out, all tasks cancelled")

asyncio.run(main())

When the timeout expires, asyncio.timeout raises TimeoutError inside the async with block. The TaskGroup catches that exception, cancels all child tasks, and then re-raises the timeout. This gives you a clean way to enforce deadlines on a group of concurrent operations.

This pattern is especially useful in server applications where you want to limit how long a request handler spends waiting for multiple services. The combination of TaskGroup and asyncio.timeout provides predictable cancellation and error propagation, making it easier to reason about resource usage and cleanup.