Python asyncio TaskGroup for concurrent operations

import asyncio
from typing import Any

# asyncio.gather (old pattern)
async def fetch_all_gather(ids: list[str]) -> list[dict]:
    results = await asyncio.gather(
        *[fetch_trace(id) for id in ids],
        return_exceptions=True  # Gather swallows errors by default
    )
    # Have to check each result manually
    return [r for r in results if not isinstance(r, Exception)]

# TaskGroup (Python 3.11+) — preferred
async def fetch_all_taskgroup(ids: list[str]) -> list[dict]:
    results = []
    # If ANY task raises, all others are cancelled immediately
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch_trace(id)) for id in ids]
    # All tasks completed successfully here
    return [task.result() for task in tasks]

# Real example: parallel database + Redis operations
async def get_trace_with_context(trace_id: str) -> dict:
    async with asyncio.TaskGroup() as tg:
        trace_task = tg.create_task(db.get_trace(trace_id))
        votes_task = tg.create_task(db.get_votes(trace_id))
        cached_views_task = tg.create_task(redis.get(f'views:{trace_id}'))

    trace = trace_task.result()
    votes = votes_task.result()
    views = cached_views_task.result() or 0

    return {**trace.dict(), 'votes': votes, 'view_count': int(views)}

# With timeout
async def fetch_with_timeout(ids: list[str], timeout: float = 5.0) -> list[dict]:
    try:
        async with asyncio.timeout(timeout):
            async with asyncio.TaskGroup() as tg:
                tasks = [tg.create_task(fetch_trace(id)) for id in ids]
        return [t.result() for t in tasks]
    except TimeoutError:
        return []  # Partial results discarded; all tasks cancelled