sgn.errors

Public exception types raised by sgn pipelines.

Importable from the top-level package::

from sgn import PipelineError, PipelineEOSError

Note that a failed run usually re-raises the original exception from the failing element callback unchanged (see Pipeline.run); the types here cover failures the pipeline itself raises.

PipelineEOSError

Bases: PipelineError

              flowchart TD
              sgn.errors.PipelineEOSError[PipelineEOSError]
              sgn.errors.PipelineError[PipelineError]

                              sgn.errors.PipelineError --> sgn.errors.PipelineEOSError
                


              click sgn.errors.PipelineEOSError href "" "sgn.errors.PipelineEOSError"
              click sgn.errors.PipelineError href "" "sgn.errors.PipelineError"
            

The graph drained to EOS, but one or more elements ended their streams only after absorbing an unrecoverable error.

Raised by Pipeline.run() after a successful drain when any element reports an eos_error (e.g. a resource source whose reconnect budget ran out). Sinks have flushed by the time this is raised; it exists so the process still exits abnormally instead of reporting the run as a clean end of stream. Chained from one of the underlying failures; the full set is in errors.

Attributes:

Name	Type	Description
errors	dict[str, BaseException]	The underlying failure per element name.

Source code in src/sgn/errors.py

class PipelineEOSError(PipelineError):
    """The graph drained to EOS, but one or more elements ended their
    streams only after absorbing an unrecoverable error.

    Raised by ``Pipeline.run()`` after a successful drain when any
    element reports an ``eos_error`` (e.g. a resource source whose
    reconnect budget ran out). Sinks have flushed by the time this is
    raised; it exists so the process still exits abnormally instead of
    reporting the run as a clean end of stream. Chained from one of the
    underlying failures; the full set is in ``errors``.

    Attributes:
        errors (dict[str, BaseException]): The underlying failure per
            element name.
    """

    def __init__(self, errors: dict[str, BaseException]):
        self.errors = errors
        names = ", ".join(sorted(errors))
        super().__init__(
            f"pipeline reached EOS after unrecoverable element errors: {names}"
        )

PipelineError

Bases: RuntimeError

              flowchart TD
              sgn.errors.PipelineError[PipelineError]

              

              click sgn.errors.PipelineError href "" "sgn.errors.PipelineError"
            

Base class for errors raised by the pipeline itself.

Element-callback failures propagate out of Pipeline.run() as their original exception types, not as this class; catch this to handle the pipeline's own error conditions.

Source code in src/sgn/errors.py

class PipelineError(RuntimeError):
    """Base class for errors raised by the pipeline itself.

    Element-callback failures propagate out of ``Pipeline.run()`` as
    their original exception types, not as this class; catch this to
    handle the pipeline's own error conditions.
    """