pipeline.exceptions

Pipeline exception classes with structured, user-friendly error reporting.

The :class:``PipelineExecutionError`` is raised whenever the pipeline framework
catches an error during module execution.  It bundles the module context so
that the caller (FastAPI endpoint, notebook, CLI script) receives a single
structured exception — no digging through nested tracebacks required.

Design decisions
- This module does **not** call ``logfire`` or ``logging`` directly.  Logging
  is the responsibility of the caller (``_execute_module`` in ``pipeline.py``).
  The exception merely carries structured data and provides human-readable
  ``__str__`` output.
- ``__str__`` is the **developer-facing** text (module name, class, exception
  type, original message). Use it in local runs, notebooks, and logs.
  :attr:``user_message`` is the **end-user** text (the original module error
  only) for GDIM ``errmsg``.
- ``__cause__`` (the ``from e`` chain) is preserved so Python prints the full
  cause chain automatically when the exception propagates uncaught.

Classes

class pipeline.exceptions.PipelineExecutionError

Attributes

operationstr: Human-readable description of the failing operation, e.g. "executing module" or "evaluating ConditionalLink condition".
original_errorException: The root cause exception.
module_namestr | None: mname of the module where the error originated, if known.
module_classstr | None: __class__.__name__ of the module, if known. context : dict[str, Any] Arbitrary key/value pairs that provide additional debugging context (e.g. port names, parent module, loop index).
user_messagestr: str(original_error) — the message shown to GDIM users. Does not include module name, class, or exception type.

Inherits from:: Exception

Methods:

__init__(operation: str, original_error: Exception, module_name: str | None = None, module_class: str | None = None, context: dict[str, Any] | None = None) → None

as_dict() → dict[str, Any]

Returns

Any

Suitable for structured logging calls::

    logfire.exception("pipeline error", **err.as_dict())
    logger.error("pipeline error", extra=err.as_dict())

format_traceback() → str

Properties:

user_message

Original module error only, without pipeline wrapping.

Keep str(self) / as_dict() for developers and logs.