asyncio TaskGroups: Structured Concurrency That Cancels Cleanly Under Load

# ❌ Orphans when the first await raises mid-gather without shielding
import asyncio
import aiohttp

async def fetch_all_bad(urls: list[str]) -> list[object]:
 async with aiohttp.ClientSession() as session:
 tasks = [asyncio.create_task(session.get(u)) for u in urls]
 # if something else raises before await gather, tasks keep running
 return await asyncio.gather(*tasks)

# ✅ TaskGroup owns lifetime
async def fetch_all(urls: list[str], limit: int = 32) -> list[bytes]:
 sem = asyncio.Semaphore(limit)
 results: list[bytes] = [b""] * len(urls)

 async def one(i: int, url: str) -> None:
 async with sem:
 async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
 resp.raise_for_status()
 results[i] = await resp.read()

 async with aiohttp.ClientSession() as session:
 try:
 async with asyncio.TaskGroup() as tg:
 for i, url in enumerate(urls):
 tg.create_task(one(i, url))
 except* aiohttp.ClientResponseError as eg:
 # ✅ ExceptionGroup via except* — inspect eg.exceptions
 raise RuntimeError(f"{len(eg.exceptions)} upstream failures") from eg
 return results

async def write_with_lease(conn, payload: bytes) -> None:
 lease = await conn.acquire()
 try:
 await conn.execute(payload)
 finally:
 # ✅ runs on cancel — return connection to pool
 await conn.release(lease)

async def fanout_with_sla(urls: list[str]) -> list[bytes]:
 async with asyncio.timeout(2.0): # ✅ whole fan-out budget
 return await fetch_all(urls)