Exception Groups & except*
debt(d7/e3/b3/t7)
Closest to 'only careful code review or runtime testing' (d7) — mypy/ruff don't reliably flag 'only first exception caught' patterns in asyncio.gather; missed concurrent failures typically surface in code review or when failures slip through silently.
Closest to 'simple parameterised fix' (e3) — quick_fix is replacing asyncio.gather with TaskGroup and using except* blocks, a localized pattern swap within the affected async code paths.
Closest to 'localised tax' (b3) — applies primarily to async/concurrent code sections; doesn't shape the whole system but adds ongoing handling complexity wherever TaskGroup is used.
Closest to 'serious trap' (t7) — per misconception, developers assume except* is a drop-in for except; it contradicts familiar exception semantics by operating on groups and leaving unmatched exceptions to propagate.
Also Known As
TL;DR
Explanation
ExceptionGroup (Python 3.11) wraps multiple exceptions raised simultaneously. except* syntax catches specific exception types from the group while letting unmatched exceptions propagate. asyncio.TaskGroup (Python 3.11) runs all tasks and collects all failures in an ExceptionGroup — unlike asyncio.gather which only surfaces the first failure. The except* handler receives an ExceptionGroup containing only the matching exceptions.
Common Misconception
Why It Matters
Common Mistakes
- Using except* for single exceptions — it is specifically for ExceptionGroup
- Not handling remaining unmatched exceptions — they still propagate
- Forgetting asyncio.TaskGroup requires Python 3.11+ — use asyncio.gather for older versions
- Not iterating eg.exceptions to see individual exception details
Code Examples
# asyncio.gather — only first error visible:
async def run_all():
results = await asyncio.gather(task1(), task2(), task3())
# If task1 and task3 both fail:
# Only task1's error is raised — task3's error is silently lost
# asyncio.TaskGroup — all errors collected:
async def run_all():
async with asyncio.TaskGroup() as tg:
t1 = tg.create_task(task1())
t2 = tg.create_task(task2())
t3 = tg.create_task(task3())
# All tasks ran; both failures collected in ExceptionGroup
try:
await run_all()
except* ValueError as eg:
for exc in eg.exceptions:
print(f'ValueError from task: {exc}')
except* ConnectionError as eg:
for exc in eg.exceptions:
print(f'Connection failed: {exc}')