Python⚡Concurrencyintermediate

`asyncio.TaskGroup` and `ExceptionGroup` (Python 3.11+)

Python · Concurrency · Run with python3

Explanation

`asyncio.TaskGroup` (PEP 654) is the structured concurrency primitive for asyncio: all tasks in a group are cancelled if one raises, and multiple exceptions are bundled into an `ExceptionGroup`. `except*` syntax handles specific exception types from a group, ignoring others. This replaces the fragile pattern of manually cancelling tasks on failure and re-raising the first exception — common bugs with bare `gather()`.

Real-World Scenario

Fetching data from 5 microservices simultaneously: with `gather`, one failure means you either get all results (if `return_exceptions=True`) or only the first exception. With `TaskGroup`, all in-flight requests are cancelled immediately on the first failure, and all exceptions are collected — no requests are left running without a consumer. ```python import asyncio import random async def fetch_service(name: str, fail: bool = False) -> str: await asyncio.sleep(random.uniform(0.05, 0.2)) if fail: raise ValueError(f"{name} returned an error") return f"{name}: OK" async def main() -> None: # TaskGroup: structured concurrency (Python 3.11+) try: async with asyncio.TaskGroup() as tg: t1 = tg.create_task(fetch_service("auth")) t2 = tg.create_task(fetch_service("inventory")) t3 = tg.create_task(fetch_service("pricing", fail=True)) except* ValueError as eg: print(f"Some services failed: {[str(e) for e in eg.exceptions]}") return print([t1.result(), t2.result(), t3.result()]) asyncio.run(main()) ```

Quick Snippet

snippet.py

python

import asyncio
import random

async def fetch_service(name: str, fail: bool = False) -> str:
    await asyncio.sleep(random.uniform(0.05, 0.2))
    if fail:
        raise ValueError(f"{name} returned an error")
    return f"{name}: OK"

async def main() -> None:
    # TaskGroup: structured concurrency (Python 3.11+)
    try:
        async with asyncio.TaskGroup() as tg:
            t1 = tg.create_task(fetch_service("auth"))
            t2 = tg.create_task(fetch_service("inventory"))
            t3 = tg.create_task(fetch_service("pricing", fail=True))
    except* ValueError as eg:
        print(f"Some services failed: {[str(e) for e in eg.exceptions]}")
        return

    print([t1.result(), t2.result(), t3.result()])

asyncio.run(main())

Full Example

View on GitHub

task_group.py

python

# Compile: python3 task_group.py
# Output:
# Some services failed (Python 3.11+) or error caught

import asyncio
import sys
import random

async def fetch_service(name: str, fail: bool = False) -> str:
    await asyncio.sleep(random.uniform(0.01, 0.05))
    if fail:
        raise ValueError(f"{name} returned an error")
    return f"{name}: OK"

async def main_with_task_group() -> None:
    if sys.version_info < (3, 11):
        print("TaskGroup requires Python 3.11+")
        print(f"Running Python {sys.version_info.major}.{sys.version_info.minor}")
        return

    try:
        async with asyncio.TaskGroup() as tg:
            t1 = tg.create_task(fetch_service("auth"))
            t2 = tg.create_task(fetch_service("inventory"))
            t3 = tg.create_task(fetch_service("pricing", fail=True))
    except* ValueError as eg:
        print(f"Some services failed: {[str(e) for e in eg.exceptions]}")
    else:
        print([t1.result(), t2.result(), t3.result()])

async def main_with_gather() -> None:
    results = await asyncio.gather(
        fetch_service("auth"),
        fetch_service("inventory"),
        fetch_service("pricing", fail=True),
        return_exceptions=True,
    )
    for r in results:
        if isinstance(r, Exception):
            print(f"Error: {r}")
        else:
            print(r)

print("=== TaskGroup (3.11+) ===")
asyncio.run(main_with_task_group())
print("\n=== gather with return_exceptions ===")
asyncio.run(main_with_gather())

Run command

bash

python3 task_group.py

Performance Patterns — Choosing the Right Concurrency Model

Python

Metaclasses

Python