Managed Concurrency with TaskGroups

Snippet

Managed Concurrency with TaskGroups

Introduced in Python 3.11, TaskGroups provide a modern API for concurrent tasks. If one task fails, the others are automatically cancelled, ensuring an 'all-or-nothing' execution pattern with clean exception handling via ExceptionGroups.

snippet.py
python
import asyncio
 
async def fail_task():
    await asyncio.sleep(0.1)
    raise ValueError("Atomic failure")
 
async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(asyncio.sleep(1))
            tg.create_task(fail_task())
    except* ValueError as eg:
        for e in eg.exceptions:
            print(f"Caught: {e}")

Breakdown

async with asyncio.TaskGroup() as tg:

Creates a scope that waits for all tasks to complete before exiting.

except* ValueError as eg:

Uses the new 'except*' syntax to handle multiple exceptions wrapped in an ExceptionGroup.

Previous snippet

From your library

Managed Concurrency with TaskGroups

Related