Exception Groups and the except* Operator

Exception Groups and the except* Operator

After reading this chapter, you will master Python’s concurrent error-handling architecture introduced in PEP 654, construct and inspect hierarchical ExceptionGroup trees, handle multiple simultaneous failures using the except* clause, isolate specific error sub-types, and prevent mixed-handler syntax traps.

Mental model

In traditional sequential programming, execution halts at the first raised exception. However, in concurrent environments (such as thread pools, worker queues, and asyncio.TaskGroup), multiple tasks execute in parallel and can fail simultaneously.

Prior to Python 3.11, runtimes were forced to discard all but the first failure or bury errors in custom aggregate lists. ExceptionGroup and the except* operator provide first-class language support for handling multiple independent exceptions at once:

Raised ExceptionGroup:
  ExceptionGroup("Batch deployment failure", [
      ConnectionError("Host 10.0.1.1 unreachable"),
      TimeoutError("Host 10.0.1.2 timeout (30s)"),
      ConnectionError("Host 10.0.1.3 connection reset")
  ])
                    │
                    ▼
          [ try ... except* ]
                    │
  ├─ except* ConnectionError as eg:
  │    │
  │    └─ Matched Subgroup: [ ConnectionError(10.0.1.1), ConnectionError(10.0.1.3) ]
  │       Executes this block!
  │
  └─ except* TimeoutError as eg:
       │
       └─ Matched Subgroup: [ TimeoutError(10.0.1.2) ]
          ALSO executes this block!

Traditional except vs Modern except*

┌───────────────────────────────────────┬───────────────────────────────────────┐
│ Feature                               │ Traditional except                    │ Modern except*                        │
├───────────────────────────────────────┼───────────────────────────────────────┤
│ Handled exceptions                    │ Exactly one per block                 │ Multiple exceptions in a group        │
│ Clause execution                      │ First matching clause wins; stops     │ Multiple matching clauses can execute │
│ Unmatched exceptions                  │ Replaces raised error                 │ Automatically re-raised in group      │
│ Target type                           │ Any BaseException                     │ Subsets of ExceptionGroup             │
└───────────────────────────────────────┴───────────────────────────────────────┘

---

Minimal example

Save as exception_groups_demo.py:

# exception_groups_demo.py
def deploy_microservices() -> None:
    """Simulate parallel deployment tasks where multiple failures occur."""
    failures = [
        ConnectionError("Failed to connect to redis-01.internal:6379"),
        TimeoutError("Database migration timed out after 60s"),
        ConnectionError("Failed to connect to redis-02.internal:6379"),
    ]
    raise ExceptionGroup("Microservice deployment failed", failures)

def main() -> None:
    try:
        deploy_microservices()
    except* ConnectionError as eg:
        # Matches and extracts ONLY the ConnectionError instances
        print(f"[RECOVER] Handling {len(eg.exceptions)} connection failures:")
        for exc in eg.exceptions:
            print(f"  - {exc}")
    except* TimeoutError as eg:
        # Matches and extracts ONLY the TimeoutError instances in the SAME group!
        print(f"[ALERT] Escalating {len(eg.exceptions)} timeout failures:")
        for exc in eg.exceptions:
            print(f"  - {exc}")

if __name__ == "__main__":
    main()

Run via uv run python exception_groups_demo.py:

[RECOVER] Handling 2 connection failures:
  - Failed to connect to redis-01.internal:6379
  - Failed to connect to redis-02.internal:6379
[ALERT] Escalating 1 timeout failures:
  - Database migration timed out after 60s

Notice: Both except* clauses executed during the single try evaluation!

---

Worked examples

Case 1: Concurrent Node Health Probing with Diagnostic Subgroups

When monitoring a distributed cluster, diagnostic probes execute across dozens of servers simultaneously. Handling failures requires distinguishing transient network errors from permanent authentication rejections:

# cluster_probe_monitor.py
class AuthRejectedError(Exception):
    """Raised when node TLS certificate or API token is invalid."""

def simulate_cluster_audit() -> None:
    errors = [
        TimeoutError("Node 10.0.2.14: Health probe response exceeded 5000ms"),
        AuthRejectedError("Node 10.0.2.18: Client certificate expired"),
        TimeoutError("Node 10.0.2.22: Health probe response exceeded 5000ms"),
    ]
    raise ExceptionGroup("Cluster audit uncovered node health anomalies", errors)

def main() -> None:
    try:
        simulate_cluster_audit()
    except* TimeoutError as eg:
        print(f"[AUTO-RETRY] Enqueueing {len(eg.exceptions)} nodes for exponential backoff:")
        for err in eg.exceptions:
            print(f"  Transient: {err}")
    except* AuthRejectedError as eg:
        print(f"[SECURITY ALERT] Page on-call immediately for {len(eg.exceptions)} security faults:")
        for err in eg.exceptions:
            print(f"  Fatal: {err}")

if __name__ == "__main__":
    main()

Run:

uv run python cluster_probe_monitor.py

Output:

[AUTO-RETRY] Enqueueing 2 nodes for exponential backoff:
  Transient: Node 10.0.2.14: Health probe response exceeded 5000ms
  Transient: Node 10.0.2.22: Health probe response exceeded 5000ms
[SECURITY ALERT] Page on-call immediately for 1 security faults:
  Fatal: Node 10.0.2.18: Client certificate expired

---

Case 2: Programmatic Group Splitting and Filtering (.split() and .subgroup())

In addition to except* syntax, ExceptionGroup instances provide programmatic methods to filter and inspect exception trees directly:

# group_introspection.py
def main() -> None:
    group = ExceptionGroup("Data pipeline failure", [
        ValueError("Column 'age' cannot be negative"),
        TypeError("Expected float for column 'latency', got str"),
        ValueError("Column 'score' exceeds maximum 100"),
        RuntimeError("Database write lock acquisition failed"),
    ])

    # 1. .split(predicate) divides the group into (matching_subgroup, remaining_group)
    value_errors, rest = group.split(ValueError)

    print("--- Extracted ValueError Subgroup ---")
    if value_errors:
        for exc in value_errors.exceptions:
            print(f"  Validation issue: {exc}")

    print("\n--- Remaining Unmatched Group ---")
    if rest:
        for exc in rest.exceptions:
            print(f"  System issue:     {exc}")

    # 2. .subgroup(predicate) filters matching items recursively
    type_subgroup = group.subgroup(TypeError)
    print(f"\nType error subgroup item count: {len(type_subgroup.exceptions) if type_subgroup else 0}")

if __name__ == "__main__":
    main()

Run:

uv run python group_introspection.py

Output:

--- Extracted ValueError Subgroup ---
  Validation issue: Column 'age' cannot be negative
  Validation issue: Column 'score' exceeds maximum 100

--- Remaining Unmatched Group ---
  System issue:     Expected float for column 'latency', got str
  System issue:     Database write lock acquisition failed

Type error subgroup item count: 1

---

Case 3: Enriching Individual Errors in Groups with add_note()

Python 3.11+ allows attaching contextual notes to any exception via exc.add_note("text"). When combined with ExceptionGroup, notes provide diagnostics for individual concurrent branches without mutating original exception messages:

# diagnostic_notes.py
def perform_shard_sync() -> None:
    e1 = ConnectionResetError("Remote server terminated socket")
    e1.add_note("Shard ID: shard-001 (us-east-1)")
    e1.add_note("Retry attempt: 3 of 3 exhausted")

    e2 = PermissionError("Access denied to replication bucket")
    e2.add_note("Shard ID: shard-004 (eu-central-1)")
    e2.add_note("IAM Role: arn:aws:iam::123456789012:role/StorageReader")

    raise ExceptionGroup("Shard replication pipeline failed", [e1, e2])

def main() -> None:
    try:
        perform_shard_sync()
    except* Exception as eg:
        for exc in eg.exceptions:
            print(f"\nCaught: {type(exc).__name__}: {exc}")
            if hasattr(exc, "__notes__"):
                print("Contextual Notes:")
                for note in exc.__notes__:
                    print(f"  * {note}")

if __name__ == "__main__":
    main()

Run:

uv run python diagnostic_notes.py

Output:

Caught: ConnectionResetError: Remote server terminated socket
Contextual Notes:
  * Shard ID: shard-001 (us-east-1)
  * Retry attempt: 3 of 3 exhausted

Caught: PermissionError: Access denied to replication bucket
Contextual Notes:
  * Shard ID: shard-004 (eu-central-1)
  * IAM Role: arn:aws:iam::123456789012:role/StorageReader

---

Pitfalls

Pitfall 1: Mixing except and except* in the Same try Block

A try statement cannot mix traditional except clauses with except* clauses. Doing so results in an immediate SyntaxError:

# THE TRAP: SyntaxError
try:
    raise ExceptionGroup("errors", [ValueError("bad")])
except ValueError:     # BUG: SyntaxError: cannot have both 'except' and 'except*' on the same 'try'
    pass
except* TypeError:
    pass

# THE FIX: Use except* exclusively when handling ExceptionGroup hierarchies
try:
    raise ExceptionGroup("errors", [ValueError("bad")])
except* ValueError:
    print("Handled ValueError cleanly")
except* TypeError:
    print("Handled TypeError cleanly")

---

Pitfall 2: Partial Matching Leaves Unhandled Errors to Propagate

When except* matches a subgroup, any unmatched exceptions in the ExceptionGroup are automatically re-raised at the end of the try ... except* statement:

# THE TRAP: Unhandled exceptions escape!
def run_batch():
    raise ExceptionGroup("batch", [ValueError("invalid input"), KeyError("missing key")])

try:
    try:
        run_batch()
    except* ValueError:
        print("Handled ValueError!")
        # KeyError is NOT handled here! It is automatically re-raised as an ExceptionGroup!
except ExceptionGroup as outer_eg:
    print(f"Caught unhandled residual group: {outer_eg.exceptions}")

Output:

Handled ValueError!
Caught unhandled residual group: (KeyError('missing key'),)

---

Pitfall 3: Catching ExceptionGroup Directly with Traditional except

Catching except ExceptionGroup as eg: works, but treats the group as a single monolithic block without destructuring. You must manually inspect .exceptions instead of letting the VM dispatch by type:

# Clunky manual dispatch:
try:
    raise ExceptionGroup("test", [ValueError("v"), TypeError("t")])
except ExceptionGroup as eg:
    for e in eg.exceptions:
        if isinstance(e, ValueError): ...

# Modern idiomatic dispatch:
try:
    raise ExceptionGroup("test", [ValueError("v"), TypeError("t")])
except* ValueError as eg:
    print(f"Handled value errors: {eg.exceptions}")
except* TypeError as eg:
    print(f"Handled type errors: {eg.exceptions}")

---

Exercises

1. Create an ExceptionGroup containing two ValueError exceptions and one KeyError. Write a try ... except* block that catches only the ValueError exceptions, observing that the KeyError propagates.
2. Given an ExceptionGroup with nested subgroups, use .split() to extract all network-related exceptions (ConnectionError, TimeoutError) while leaving other errors intact.
3. Write a simulation of 5 concurrent worker tasks using a list of callables. If any fail, collect all exceptions into an ExceptionGroup and raise it.
4. Attach diagnostic notes (.add_note()) containing timestamps and hostnames to exceptions before grouping them into an ExceptionGroup.
5. Demonstrate why except* Exception will catch all standard exceptions in a group, but will still let KeyboardInterrupt and SystemExit (subclasses of BaseExceptionGroup) pass through.

---

Further reading

• PEP 654: Exception Groups and except*.
• Python Documentation: Built-in Exceptions – ExceptionGroup and BaseExceptionGroup.
• Python Documentation: The try statement – Handling Exception Groups with except*.