Structured Concurrency with TaskGroup

Why TaskGroup?

asyncio.gather is useful, but it can make lifecycle and error handling messy.

asyncio.TaskGroup (Python 3.11+) provides structured concurrency:

tasks are scoped to a block
if one task fails, the group cancels the rest
errors are grouped in an ExceptionGroup

Example

taskgroup_basic.py

import asyncio
 
 
async def job(name: str, delay: float):
    await asyncio.sleep(delay)
    return f"{name} done"
 
 
async def main():
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(job("A", 0.2))
        t2 = tg.create_task(job("B", 0.1))
 
    # task results are available after the block
    print(t1.result())
    print(t2.result())
 
 
asyncio.run(main())

Handling failures

taskgroup_error.py

import asyncio
 
 
async def ok():
    await asyncio.sleep(0.1)
    return 1
 
 
async def bad():
    await asyncio.sleep(0.05)
    raise ValueError("boom")
 
 
async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(ok())
            tg.create_task(bad())
    except* ValueError as eg:
        # ExceptionGroup support (PEP 654)
        print("handled:", eg)
 
 
asyncio.run(main())

sketch When one task fails: gather leaves the others running, TaskGroup does not p5.js

Three workers; the middle one raises. Both forms report the error, but they leave the program in different states. gather returns the exception and the surviving sibling carries on to completion -- checked afterwards, it reports done and not cancelled. TaskGroup cancels every sibling before it re-raises, wrapped in an ExceptionGroup. That is what structured concurrency means: the block does not finish while anything it started is still running.

Structured Concurrency with TaskGroup

Why TaskGroup?

Example

Handling failures

Notes

🧪 Try It Yourself

Exercise 1 – Create a Task

Exercise 2 – Gather Multiple Tasks

Exercise 3 – Task Name and Done

pch.feedbackHeading