Warning and Error Context

The warning_error_context helper adds structured context to warnings and exceptions without changing the code that emits them. This is useful when a low-level warning or error does not know which dataset item, optimization trial, or iteration triggered it.

This example covers fixed and dynamic context, nested contexts, retained event records, contextual prints, manual lifecycle control, quiet record-only execution, restoration in parallel workers, arbitrary iterators, TypedIterator, and typed sub-iterations.

Simple and dynamic context

Pass fixed values as a dictionary. Warning messages retain their original warning type and source location. The original message is followed by a contextual copy on a separate line.

import warnings
from collections.abc import Iterable, Iterator
from dataclasses import dataclass

from tpcp.misc import (
    BaseTypedIterator,
    TypedIterator,
    iter_with_warning_error_context,
    print_with_context,
    warning_error_context,
)
from tpcp.parallel import Parallel, delayed

with warning_error_context("record", {"participant": "p1"}):
    warnings.warn("signal is short", UserWarning, stacklevel=1)

/home/docs/checkouts/readthedocs.org/user_builds/tpcp/checkouts/v2.3.0/examples/recipies/_07_warning_error_context.py:37: UserWarning: signal is short
[record: participant='p1'] signal is short
  warnings.warn("signal is short", UserWarning, stacklevel=1)

A context_provider is evaluated only when a warning or exception is rendered. It can be used without passing an empty context dictionary. This is useful for state that changes while the context is active.

state = {"step": 0}

with warning_error_context(
    "processing",
    context_provider=lambda: {"step": state["step"]},
):
    state["step"] = 2
    warnings.warn("processing warning", UserWarning, stacklevel=1)

/home/docs/checkouts/readthedocs.org/user_builds/tpcp/checkouts/v2.3.0/examples/recipies/_07_warning_error_context.py:50: UserWarning: processing warning
[processing: step=2] processing warning
  warnings.warn("processing warning", UserWarning, stacklevel=1)

Fixed context and a provider can also be combined. Provider keys must not duplicate keys from the fixed dictionary.

with warning_error_context(
    "processing",
    {"participant": "p1"},
    context_provider=lambda: {"step": state["step"]},
):
    warnings.warn("combined warning", UserWarning, stacklevel=1)

/home/docs/checkouts/readthedocs.org/user_builds/tpcp/checkouts/v2.3.0/examples/recipies/_07_warning_error_context.py:60: UserWarning: combined warning
[processing: participant='p1', step=2] combined warning
  warnings.warn("combined warning", UserWarning, stacklevel=1)

Nested contexts

Contexts compose from outermost to innermost. Each context is removed when its with block ends.

with warning_error_context("participant", {"id": "p1"}):  # noqa: SIM117 - Python 3.9 compatibility
    with warning_error_context("recording", {"id": 3}):
        warnings.warn("nested warning", UserWarning, stacklevel=1)

/home/docs/checkouts/readthedocs.org/user_builds/tpcp/checkouts/v2.3.0/examples/recipies/_07_warning_error_context.py:69: UserWarning: nested warning
[participant: id='p1' > recording: id=3] nested warning
  warnings.warn("nested warning", UserWarning, stacklevel=1)

Errors

The context manager re-raises the original error, so it can be caught and handled normally. The active context is attached to the caught exception through exception notes. On Python 3.11 and newer, the standard traceback renderer displays these notes. On Python 3.9 and 3.10, renderers with exception-note support, such as Rich, can display them.

try:
    with warning_error_context("recording", {"id": 3}):
        raise ValueError("invalid signal")
except ValueError as error:
    print(f"Caught error: {error}")
    print(f"Context: {getattr(error, '__notes__', [])}")

Caught error: invalid signal
Context: ['Context: recording: id=3']

Manual lifecycle

For a long top-level script, the object returned by warning_error_context can be started and stopped explicitly to avoid indenting the entire guarded section.

manual_context = warning_error_context("script", {"phase": "processing"})
manual_context.start()
warnings.warn("manual warning", UserWarning, stacklevel=1)
print_with_context("Manual context is active")
manual_context.stop()

print([record.type for record in manual_context.records])

/home/docs/checkouts/readthedocs.org/user_builds/tpcp/checkouts/v2.3.0/examples/recipies/_07_warning_error_context.py:94: UserWarning: manual warning
[script: phase='processing'] manual warning
  warnings.warn("manual warning", UserWarning, stacklevel=1)
[script: phase='processing'] Manual context is active
['warning', 'print']

Manual lifecycle control has an important exception caveat. stop() does not receive an active exception, so it cannot record or annotate one. If an exception prevents stop() from running, the context also remains active. A try/finally can guarantee cleanup, but the bare stop() call still cannot add exception context. Use the with form whenever exceptions must be recorded, annotated, or cleanly unwind the context.

Recording events and contextual prints

Every context manager yields a persistent result whose records list can be inspected after the context exits. Each named-tuple record contains the event type ("warning", "error", or "print"), the fully resolved context stack, and the original warning/exception object or print message. Nested events are included in every active context result.

print_with_context accepts the normal print arguments. With an active context, it prepends the rendered context and adds a record. Without an active context, it behaves like a normal print.

items = ["ok", "short", "invalid"]

with warnings.catch_warnings():
    # Warning filters run before tpcp can record a warning. Use ``always`` when
    # every repeated occurrence is relevant to the later analysis.
    warnings.simplefilter("always")
    with warning_error_context(
        "dataset", {"name": "validation"}
    ) as dataset_context:
        for make_context, item in iter_with_warning_error_context(items):
            try:
                with make_context("datapoint", {"value": item}):
                    print_with_context("Processing", item)
                    if item != "ok":
                        warnings.warn(
                            "signal quality issue", UserWarning, stacklevel=1
                        )
                    if item == "invalid":
                        raise ValueError("invalid signal")
            except ValueError:
                # Errors are recorded when they leave a context. They still
                # propagate normally and can be handled by the application.
                pass

print([record.type for record in dataset_context.records])

warning_contexts = [
    record.context
    for record in dataset_context.records
    if record.type == "warning" and isinstance(record.message, UserWarning)
]
print(warning_contexts)

[dataset: name='validation' > datapoint: i=0, value='ok'] Processing ok
[dataset: name='validation' > datapoint: i=1, value='short'] Processing short
/home/docs/checkouts/readthedocs.org/user_builds/tpcp/checkouts/v2.3.0/examples/recipies/_07_warning_error_context.py:134: UserWarning: signal quality issue
[dataset: name='validation' > datapoint: i=1, value='short'] signal quality issue
  warnings.warn(
[dataset: name='validation' > datapoint: i=2, value='invalid'] Processing invalid
/home/docs/checkouts/readthedocs.org/user_builds/tpcp/checkouts/v2.3.0/examples/recipies/_07_warning_error_context.py:134: UserWarning: signal quality issue
[dataset: name='validation' > datapoint: i=2, value='invalid'] signal quality issue
  warnings.warn(
['print', 'print', 'warning', 'print', 'warning', 'error']
["dataset: name='validation' > datapoint: i=1, value='short'", "dataset: name='validation' > datapoint: i=2, value='invalid'"]

Quiet record-only mode

A high-level record_only=True context suppresses contextual warning and print_with_context output from all nested contexts while retaining the same records. It does not suppress errors or ordinary print calls. This is useful for wrapping a long-running program and printing only a final summary.

with warnings.catch_warnings():
    warnings.simplefilter("always")
    with warning_error_context(
        "program", {"mode": "batch"}, record_only=True
    ) as program_context:
        for make_context, item in iter_with_warning_error_context(
            ["left", "right"]
        ):
            with make_context("datapoint", {"value": item}):
                warnings.warn("quiet warning", UserWarning, stacklevel=1)
                print_with_context("Processed", item)

print([record.type for record in program_context.records])

['warning', 'print', 'warning', 'print']

Parallel workers

tpcp.parallel.delayed captures the active context when a task is created and restores it around that task in a process worker. Contexts opened inside the worker stack on top of the restored context. Pairing delayed with tpcp.parallel.Parallel also merges warning and contextual-print records from successful tasks back into the original parent context.

Both TPCP helpers are required for this complete round trip: tpcp.parallel.delayed restores the context in the worker, while tpcp.parallel.Parallel recovers its records in the parent. The corresponding joblib helpers do not provide both parts automatically.

def run_parallel_example():
    """Run the example without making the worker import this gallery module."""

    def process_in_worker(item: str) -> str:
        with warning_error_context("worker", {"item": item}):
            warnings.warn("worker warning", UserWarning, stacklevel=1)
            print_with_context("Worker processed", item)
        return item.upper()

    with warnings.catch_warnings():
        warnings.simplefilter("always")
        with warning_error_context(
            "parent", {"run": "validation"}, record_only=True
        ) as parallel_context:
            parallel_results = Parallel(n_jobs=2)(
                delayed(process_in_worker)(item) for item in ["left", "right"]
            )
    return parallel_results, parallel_context.records


parallel_results, parallel_records = run_parallel_example()

print(parallel_results)
print(
    [
        (record.type, record.context, str(record.message))
        for record in parallel_records
    ]
)

['LEFT', 'RIGHT']
[('warning', "parent: run='validation' > worker: item='left'", 'worker warning'), ('print', "parent: run='validation' > worker: item='left'", 'Worker processed left'), ('warning', "parent: run='validation' > worker: item='right'", 'worker warning'), ('print', "parent: run='validation' > worker: item='right'", 'Worker processed right')]

Arbitrary iterators

iter_with_warning_error_context pairs each item with a context creator. Enter the returned context explicitly inside the loop body. The creator automatically adds the zero-based iteration index as i; all other values remain explicit.

Python normally suppresses repeated warnings from the same source line. This filtering happens before context is attached, so different context values do not make the warnings distinct. Applications that need every contextual occurrence can enable an appropriate warning filter. Here, catch_warnings only scopes that filter; warnings still emit normally for Sphinx-Gallery.

with warnings.catch_warnings():
    warnings.simplefilter("always")
    for make_context, item in iter_with_warning_error_context(
        ["left", "right"]
    ):
        with make_context("item", {"value": item}):
            warnings.warn("iterator warning", UserWarning, stacklevel=1)

/home/docs/checkouts/readthedocs.org/user_builds/tpcp/checkouts/v2.3.0/examples/recipies/_07_warning_error_context.py:237: UserWarning: iterator warning
[item: i=0, value='left'] iterator warning
  warnings.warn("iterator warning", UserWarning, stacklevel=1)
/home/docs/checkouts/readthedocs.org/user_builds/tpcp/checkouts/v2.3.0/examples/recipies/_07_warning_error_context.py:237: UserWarning: iterator warning
[item: i=1, value='right'] iterator warning
  warnings.warn("iterator warning", UserWarning, stacklevel=1)

Combining regular and iterator contexts

An iterator context composes with any regular context that is already active. The regular context is rendered first, followed by the per-iteration context and its automatically injected i.

with warnings.catch_warnings():
    warnings.simplefilter("always")
    with warning_error_context("dataset", {"name": "validation"}):
        for make_context, item in iter_with_warning_error_context(
            ["left", "right"]
        ):
            with make_context("item", {"value": item}):
                warnings.warn("stacked warning", UserWarning, stacklevel=1)

/home/docs/checkouts/readthedocs.org/user_builds/tpcp/checkouts/v2.3.0/examples/recipies/_07_warning_error_context.py:252: UserWarning: stacked warning
[dataset: name='validation' > item: i=0, value='left'] stacked warning
  warnings.warn("stacked warning", UserWarning, stacklevel=1)
/home/docs/checkouts/readthedocs.org/user_builds/tpcp/checkouts/v2.3.0/examples/recipies/_07_warning_error_context.py:252: UserWarning: stacked warning
[dataset: name='validation' > item: i=1, value='right'] stacked warning
  warnings.warn("stacked warning", UserWarning, stacklevel=1)

TypedIterator

TypedIterator exposes the same context-creation interface. While an item is yielded, warning_error_context reads the current zero-based index from the iterator and injects it as i. No other item metadata is inferred.

@dataclass
class Result:
    """Result created for one iteration."""

    doubled: int


typed_iterator = TypedIterator[int, Result](Result)

with warnings.catch_warnings():
    warnings.simplefilter("always")
    for item, result in typed_iterator.iterate([2, 4]):
        with typed_iterator.warning_error_context("item", {"value": item}):
            warnings.warn("typed warning", UserWarning, stacklevel=1)
            result.doubled = item * 2

print(typed_iterator.results_.doubled)

/home/docs/checkouts/readthedocs.org/user_builds/tpcp/checkouts/v2.3.0/examples/recipies/_07_warning_error_context.py:276: UserWarning: typed warning
[item: i=0, value=2] typed warning
  warnings.warn("typed warning", UserWarning, stacklevel=1)
/home/docs/checkouts/readthedocs.org/user_builds/tpcp/checkouts/v2.3.0/examples/recipies/_07_warning_error_context.py:276: UserWarning: typed warning
[item: i=1, value=4] typed warning
  warnings.warn("typed warning", UserWarning, stacklevel=1)
[4, 8]

TypedIterator sub-iterations

Custom typed iterators can expose sub-iterations by wrapping the protected _iterate helper. The same warning_error_context method then binds to whichever iteration is currently yielding. Every sub-iteration starts its own i at zero, and completing it restores the outer iteration’s i.

class NestedIterator(BaseTypedIterator[int, Result]):
    """Typed iterator with an outer iteration and an explicit sub-iteration."""

    def iterate(self, values: Iterable[int]) -> Iterator[tuple[int, Result]]:
        """Iterate over outer values."""
        yield from self._iterate(values)

    def iterate_subitems(
        self, values: Iterable[int]
    ) -> Iterator[tuple[int, Result]]:
        """Iterate over values belonging to the current outer item."""
        yield from self._iterate(values, iteration_name="subitems")


nested_iterator = NestedIterator(Result)

with warnings.catch_warnings():
    warnings.simplefilter("always")
    for outer, outer_result in nested_iterator.iterate([10]):
        with nested_iterator.warning_error_context("outer", {"value": outer}):
            warnings.warn("outer before", UserWarning, stacklevel=1)

        for inner, inner_result in nested_iterator.iterate_subitems([20, 21]):
            with nested_iterator.warning_error_context(
                "inner",
                context_provider=lambda inner=inner: {"value": inner},
            ):
                warnings.warn("inner", UserWarning, stacklevel=1)
                inner_result.doubled = inner * 2

        # No fixed dictionary is required. This context contains only the
        # restored outer ``i``.
        with nested_iterator.warning_error_context("outer_after"):
            warnings.warn("outer after", UserWarning, stacklevel=1)
            outer_result.doubled = outer * 2

/home/docs/checkouts/readthedocs.org/user_builds/tpcp/checkouts/v2.3.0/examples/recipies/_07_warning_error_context.py:310: UserWarning: outer before
[outer: i=0, value=10] outer before
  warnings.warn("outer before", UserWarning, stacklevel=1)
/home/docs/checkouts/readthedocs.org/user_builds/tpcp/checkouts/v2.3.0/examples/recipies/_07_warning_error_context.py:317: UserWarning: inner
[inner: i=0, value=20] inner
  warnings.warn("inner", UserWarning, stacklevel=1)
/home/docs/checkouts/readthedocs.org/user_builds/tpcp/checkouts/v2.3.0/examples/recipies/_07_warning_error_context.py:317: UserWarning: inner
[inner: i=1, value=21] inner
  warnings.warn("inner", UserWarning, stacklevel=1)
/home/docs/checkouts/readthedocs.org/user_builds/tpcp/checkouts/v2.3.0/examples/recipies/_07_warning_error_context.py:323: UserWarning: outer after
[outer_after: i=0] outer after
  warnings.warn("outer after", UserWarning, stacklevel=1)

In all iterator variants, the context manager is created and entered inside the loop body. This keeps its lifetime explicit and prevents context from leaking across a generator’s yield boundary.

Estimated memory usage: 8 MB