If you've spent any time with asyncio or are starting to use TaskGroup in Python 3.11+, you've likely hit a wall where your error handling suddenly stops working. You know the feeling: you wrote a perfectly good try/except block for a specific error, but Python is ignoring it and crashing your program anyway.

Let's look at a piece of code that looks correct on the surface but fails in production.

import asyncio

async def fetch_data(id):
    if id == 1:
        raise ValueError("Invalid ID")
    if id == 2:
        raise TypeError("Wrong type")
    return f"Data {id}"

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fetch_data(1))
            tg.create_task(fetch_data(2))
    except ValueError:
        print("Caught the ValueError!")

asyncio.run(main())

The invisible wall of ExceptionGroup

If you run this, you'll notice something frustrating: "Caught the ValueError!" is never printed. Instead, the program crashes with a massive traceback showing an ExceptionGroup containing both a ValueError and a TypeError.

Here is why this is happening: when multiple tasks in a TaskGroup fail, Python doesn't just pick one exception to throw. It bundles all of them into an ExceptionGroup. This is a wrapper. Because the ExceptionGroup itself is not a ValueError, your except ValueError: block is completely bypassed. It's like trying to find a specific book by looking at the cover of a cardboard box; the book is inside, but the box itself isn't a book.

Unpacking the group with except*

To handle this, Python introduced a new piece of syntax: except*. Think of this as "except if the group contains." It doesn't catch the group itself; it reaches inside the group and pulls out only the exceptions that match the type you specified.

import asyncio

async def fetch_data(id):
    if id == 1:
        raise ValueError("Invalid ID")
    if id == 2:
        raise TypeError("Wrong type")
    return f"Data {id}"

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fetch_data(1))
            tg.create_task(fetch_data(2))
    except* ValueError as eg:
        for e in eg.exceptions:
            print(f"Handled a value error: {e}")
    except* TypeError as eg:
        for e in eg.exceptions:
            print(f"Handled a type error: {e}")

asyncio.run(main())

Now, both blocks execute. I'll point out a weird quirk here: except* handles all matching exceptions in the group. If your group had three different ValueErrors, the first except* ValueError block would trigger once, but the eg object would contain all three of them. That's why I used a for loop inside the block.

Mixing standard except and except*

One rule you absolutely cannot break: you cannot mix except and except* in the same try statement. If you try to do that, Python will throw a SyntaxError.

You have to decide: am I dealing with a single potential failure (use except), or am I dealing with a collection of concurrent tasks where multiple things could go wrong (use except*)? In a modern asynchronous codebase, you'll find yourself leaning toward except* more often than you might expect. It's a shift in mindset—you're no longer catching a single event, but filtering a stream of failures.

📋 Practical Task

Building a Resilient Batch File Processor

You are building a system that processes a batch of files. Some files might be missing (FileNotFoundError), and some might have corrupted content (RuntimeError). Since you are using a TaskGroup to process these files concurrently, you need to handle these errors without letting one bad file crash the entire batch.

Your Task: Modify the provided code to use except*. Ensure that: 1. All FileNotFoundError exceptions are caught and printed as "File missing: [error message]". 2. All RuntimeError exceptions are caught and printed as "Corrupt file: [error message]". 3. Any other unexpected exceptions are allowed to propagate and crash the program (so don't use a catch-all Exception block).

import asyncio

async def process_file(name):
    if name == "missing.txt":
        raise FileNotFoundError(f"{name} not found on disk")
    if name == "corrupt.txt":
        raise RuntimeError(f"{name} has invalid checksum")
    print(f"Successfully processed {name}")

async def main():
    files = ["valid1.txt", "missing.txt", "corrupt.txt", "valid2.txt"]
    try:
        async with asyncio.TaskGroup() as tg:
            for f in files:
                tg.create_task(process_file(f))
    # TODO: Implement the correct except* blocks here
    
asyncio.run(main())