Structured Concurrency with TaskGroup

Why TaskGroup?

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

asyncio.TaskGroupasyncio.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 ExceptionGroupExceptionGroup

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())

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())

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())

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

Was this page helpful?