Structured Concurrency with asyncio.TaskGroup (Python 3.11+)

1Concept

Python 3.11 introduced `asyncio.TaskGroup` for structured concurrency. If any task inside the TaskGroup fails, all remaining sibling tasks are automatically cancelled immediately, and errors are packaged into an `ExceptionGroup`.

2Architecture Diagram

async with asyncio.TaskGroup() as tg:
  tg.create_task(Task A)
  tg.create_task(Task B) ---> If B throws, Task A is cancelled instantly! No orphaned tasks!

3Code Example

import asyncio

async def background_worker(name: str):
    await asyncio.sleep(0.01)
    return f"Worker {name} finished"

async def main():
    results = []
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(background_worker("A"))
        t2 = tg.create_task(background_worker("B"))
    print(f"TaskGroup completed safely: {[t1.result(), t2.result()]}")

asyncio.run(main())

4Expected Output

TaskGroup completed safely: ['Worker A finished', 'Worker B finished']

5Key Takeaways

• ✓`TaskGroup` eliminates orphaned coroutine leaks when exceptions occur.
• ✓Awaits all scheduled tasks automatically upon exiting the `async with` block.
• ✓Replaces legacy `asyncio.gather()` as the best practice in Python 3.11+.