Keep Batch Failures Typed: Build Better Python APIs with ExceptionGroup
Preserve error types, tracebacks, and context when independent operations fail together.
Batch code has an awkward failure mode. Several independent operations run, more than one fails, and the API must decide what to report. Raising the first exception hides later failures. Returning a list of strings keeps every message but throws away types and tracebacks. Logging inside the batch leaks policy into low-level code and leaves the caller with no structured result.
Python’s ExceptionGroup gives this problem a language-level answer. A group can carry multiple exceptions as one raised value. The caller can use except* to handle selected exception types while unhandled parts continue propagating. The useful design is not simply to wrap every error in a group. It is to make grouped failure an explicit API contract for operations that are independent enough to complete before the caller decides what to do.
This article targets CPython 3.14 and uses only the standard library. ExceptionGroup and except* were added in Python 3.11. The examples focus on batch validation and concurrent work, but the same contract applies to fan-out API calls, plugin execution, file imports, and multi-target automation.
Strings are a reporting format, not an error model
A common batch validator accumulates messages so it can report all bad rows at once:
from collections.abc import Iterable
def validate_rows(rows: Iterable[dict[str, object]]) -> list[str]:
errors: list[str] = []
for row_number, row in enumerate(rows, start=1):
if "email" not in row:
errors.append(f"row {row_number}: missing email")
age = row.get("age")
if isinstance(age, bool) or not isinstance(age, int) or age < 0:
errors.append(f"row {row_number}: age must be a non-negative integer")
return errors
def main() -> None:
rows: list[dict[str, object]] = [
{"age": 34},
{"email": "dev@example.com", "age": -2},
]
for message in validate_rows(rows):
print(message)
if __name__ == "__main__":
main()This program is complete and predictable, but its return type has flattened every failure into text. A caller that wants to count missing fields separately from invalid values must parse strings or duplicate validation logic. There is no traceback pointing to the condition that created each failure. Adding machine-readable context later means inventing another result schema.
A list of strings is fine at a user-interface boundary. It is a weak internal contract because display text has become both the diagnostic and the data model. The batch layer should preserve structured failures until a caller deliberately renders them.
Raise one value without erasing the individual errors
ExceptionGroup accepts a message and a sequence of exceptions. Custom exception classes can carry domain context in attributes while normal exception machinery preserves their tracebacks.
from collections.abc import Iterable
class RowValidationError(ValueError):
def __init__(self, row_number: int, message: str) -> None:
self.row_number = row_number
super().__init__(message)
class MissingFieldError(RowValidationError):
pass
class InvalidValueError(RowValidationError):
pass
def validate_rows(rows: Iterable[dict[str, object]]) -> None:
failures: list[Exception] = []
for row_number, row in enumerate(rows, start=1):
if "email" not in row:
try:
raise MissingFieldError(row_number, "missing email")
except MissingFieldError as exc:
failures.append(exc)
age = row.get("age")
if isinstance(age, bool) or not isinstance(age, int) or age < 0:
try:
raise InvalidValueError(
row_number,
"age must be a non-negative integer",
)
except InvalidValueError as exc:
failures.append(exc)
if failures:
raise ExceptionGroup("row validation failed", failures)
def main() -> None:
rows: list[dict[str, object]] = [
{"age": 34},
{"email": "dev@example.com", "age": -2},
]
try:
validate_rows(rows)
except* MissingFieldError as group:
for missing in group.exceptions:
if isinstance(missing, MissingFieldError):
print(f"missing field at row {missing.row_number}: {missing}")
except* InvalidValueError as group:
for invalid in group.exceptions:
if isinstance(invalid, InvalidValueError):
print(f"invalid value at row {invalid.row_number}: {invalid}")
if __name__ == "__main__":
main()The function has no successful return value. It either validates the complete input or raises one grouped failure. Each leaf remains a real MissingFieldError or InvalidValueError, and each carries row_number as structured context. The small raise and except around each leaf is intentional in this example: it attaches a traceback at the exact validation site before storing the exception. Constructing an exception without raising it is legal, but its __traceback__ remains None until it is raised. The isinstance() checks narrow each item for type checkers, because group.exceptions can also hold nested groups.
Both except* clauses can run for the same raised group. Python partitions the group by matching leaf type. Each clause receives a subgroup with the original hierarchy preserved, and any unmatched remainder is reraised after the clauses finish. This differs from regular except, where the first matching clause handles the whole raised object.
The tradeoff is that grouped failure is a breaking API choice. A caller written for one ordinary exception will not automatically gain useful partial handling. PEP 654 explicitly calls out that changing an existing API to raise exception groups can break callers. Introduce the contract in a new method, version, or clearly documented major change.
The traceback-preserving technique also has a cost: collecting many failures retains stack-frame references until the group is released. For very large inputs, cap the number of detailed failures and include a separate omitted count. Do not retain an unbounded group just because the language supports one.
Group failures only when operations are independent
Collecting every error is correct only if continuing after one failure is safe. Validation checks usually qualify because one bad row does not change another row. Independent reads often qualify. A sequence of dependent writes usually does not.
Imagine three steps that debit an account, write a shipment, and send a receipt. Continuing after the debit fails does not provide a more complete diagnostic. It creates an invalid workflow. That operation needs fail-fast control, compensation, or a transaction boundary, not a prettier collection of exceptions.
A practical rule is to ask whether each operation would still be valid if every sibling operation failed. If the answer is yes, complete the batch and aggregate. If one result changes whether another operation should run, preserve the dependency and stop at the appropriate boundary.
Grouping also should not replace a normal result type. Expected per-item outcomes such as “not found,” “already processed,” or a validation warning may belong in a report object. Reserve exceptions for outcomes that callers treat as failures. Otherwise, except* becomes an elaborate branch statement.
Concurrent batches need an explicit completion policy
asyncio.TaskGroup uses exception groups, but it is fail-fast: when a child fails with an exception other than CancelledError, the group cancels the remaining children. That is the right default for related tasks sharing one scope. It is not an “attempt every target” policy.
When targets are independent and all attempts should finish, capture ordinary exceptions at the item boundary, preserve cancellation, then raise one group after all tasks settle:
import asyncio
from collections.abc import Awaitable, Iterable
from dataclasses import dataclass
class TargetError(RuntimeError):
def __init__(self, target: str, message: str) -> None:
self.target = target
super().__init__(f"{target}: {message}")
@dataclass(frozen=True, slots=True)
class Target:
name: str
should_fail: bool = False
async def send(target: Target) -> None:
await asyncio.sleep(0.01)
if target.should_fail:
raise TargetError(target.name, "delivery rejected")
print(f"delivered to {target.name}")
async def capture_failure(work: Awaitable[None]) -> Exception | None:
try:
await work
except asyncio.CancelledError:
raise
except Exception as exc:
return exc
return None
async def send_to_all(targets: Iterable[Target]) -> None:
outcomes = await asyncio.gather(
*(capture_failure(send(target)) for target in targets)
)
failures = [outcome for outcome in outcomes if outcome is not None]
if failures:
raise ExceptionGroup("one or more deliveries failed", failures)
async def main() -> None:
targets = [
Target("alpha"),
Target("beta", should_fail=True),
Target("gamma", should_fail=True),
]
try:
await send_to_all(targets)
except* TargetError as group:
for error in group.exceptions:
if isinstance(error, TargetError):
print(f"failed target: {error.target}")
if __name__ == "__main__":
asyncio.run(main())capture_failure() catches Exception, not BaseException, and explicitly reraises asyncio.CancelledError. Cancellation is control flow for the surrounding async scope. Turning it into a batch leaf could make shutdown or request cancellation appear to succeed while work keeps running.
Get Udit Rawat’s stories in your inbox
Join Medium for free to get updates from this writer.
The helper returns exceptions as values only inside the implementation. The public boundary converts them back into raised, typed failures. asyncio.gather() waits for every wrapper because each ordinary item failure is captured. If the parent task is cancelled, cancellation still propagates through gather() to unfinished children. A child cancelled on its own is different: gather() raises CancelledError to the caller while the other children may keep running.
This policy differs deliberately from TaskGroup. Use TaskGroup when one child failure should cancel siblings. Use an attempt-all collector when the operations are independent and the complete failure set is useful. State that policy in the function name, type hints, and tests. Do not make callers infer it from implementation details.
Test the shape of the failure contract
Tests should assert more than “an exception happened.” Check the group message, leaf types, item context, and cancellation policy. For a flat group, subgroup() is a convenient way to inspect selected leaves without duplicating the interpreter's matching logic.
import asyncio
import unittest
class TargetError(RuntimeError):
def __init__(self, target: str) -> None:
self.target = target
super().__init__(f"{target} failed")
async def run_target(target: str) -> None:
await asyncio.sleep(0)
if target != "healthy":
raise TargetError(target)
async def run_all(targets: list[str]) -> None:
async def capture(target: str) -> Exception | None:
try:
await run_target(target)
except asyncio.CancelledError:
raise
except Exception as exc:
return exc
return None
outcomes = await asyncio.gather(*(capture(target) for target in targets))
failures = [outcome for outcome in outcomes if outcome is not None]
if failures:
raise ExceptionGroup("target batch failed", failures)
class BatchFailureTests(unittest.IsolatedAsyncioTestCase):
async def test_group_preserves_every_failed_target(self) -> None:
with self.assertRaises(ExceptionGroup) as caught:
await run_all(["healthy", "west", "east"])
group = caught.exception
target_failures = group.subgroup(TargetError)
self.assertIsNotNone(target_failures)
assert target_failures is not None
self.assertEqual(
[
error.target
for error in target_failures.exceptions
if isinstance(error, TargetError)
],
["west", "east"],
)
if __name__ == "__main__":
unittest.main()The test uses IsolatedAsyncioTestCase so the asynchronous contract is exercised directly. It proves that the healthy target does not prevent both failed targets from being represented, and that their domain context survives grouping. It does not assert a formatted traceback, which would couple the test to presentation details.
For nested groups, do not assume group.exceptions contains only leaves. It can contain child groups. Use subgroup() or split() when filtering, or write an explicit recursive walker when exporting diagnostics. Keep nesting meaningful, such as grouping first by import file and then by validation stage. Arbitrary nesting makes logs and handlers harder to understand.
Operational details still matter
Exception groups preserve information, but they do not create an observability policy. Log the group once at the boundary that owns the batch outcome. If every leaf logs itself and the parent logs the full group, one failure can appear several times. Include stable item identifiers in custom exception attributes, then let the logging or API layer decide how much detail to expose.
Do not send raw exception messages to untrusted clients. A grouped traceback may include file paths, input fragments, or library details. Translate known leaf types into safe API errors and keep diagnostic tracebacks in restricted logs. Unknown leaves should remain unhandled or map to a generic server error rather than being silently treated as expected failures.
Also define a size limit. A million invalid records should not produce a million exception objects. Stop after a configured number of detailed failures, attach an omitted count to the surrounding report, and tell the caller that validation was truncated. The cap is part of the API contract, not merely a performance optimization.
A practical contract checklist
Use ExceptionGroup when all of these are true:
- Several operations can fail independently in one logical call.
- Completing the remaining operations after one failure is safe.
- Callers benefit from handling failures by type.
- Individual tracebacks or structured item context are worth preserving.
- The API documents whether the batch is fail-fast or attempt-all.
Do not use it when operations are dependent, a transaction should stop at the first failure, expected outcomes fit a result object, or the possible group is unbounded.
When adopting the pattern, define domain-specific leaf exceptions, preserve cancellation and other control-flow exceptions, cap retained failures, test the group shape, and render messages only at the outer boundary. Most importantly, treat grouped failure as part of the public API. It changes how callers catch, retry, log, and expose errors.
ExceptionGroup is useful because it avoids choosing between the first traceback and a bag of strings. Used carefully, it keeps each failure typed and inspectable while still giving the caller one value to catch.
Sources
- Python 3.14 built-in exception groups
- Python 3.14
except*language reference - Python 3.14 asyncio task groups
- Python 3.14 asyncio gathering results
- PEP 654: Exception Groups and
except*
Udit Rawat is a Senior Software Engineer with more than 10 years of experience and co-founder of Moosky AI.
