GitHub

API reference for Cause.

Attributes

AttributeTypeDescription
reasonstuple[Reason[E_co], ...]
annotationsContext

Every reason's annotations; a later reason wins a shared key.

failurestuple[E_co, ...]
defectstuple[BaseException, ...]
interruptorsfrozenset[int]

The ids of the fibers that interrupted; external interrupts have none.

has_failsbool
has_diesbool
has_interruptsbool
has_interrupts_onlybool

Functions

__init__

__init__(reasons) -> None
ParameterTypeDescription
reasonstuple[Reason[E_co], ...]

add_interrupts

add_interrupts(interrupts) -> Cause[E_co]

Add each interruption not recorded yet: one per interrupting fiber and the frame it interrupted from.

ParameterTypeDescription
interruptsCause[Never]

Returns Cause[E_co]

def add_interrupts(self, interrupts: Cause[Never]) -> Cause[E_co]:
    """Add each interruption not recorded yet: one per interrupting
    fiber and the frame it interrupted from."""
    seen = {_interruptor(r) for r in self.reasons if isinstance(r, Interrupt)}
    new: list[Reason[E_co]] = []
    for r in interrupts.reasons:
        if isinstance(r, Interrupt) and (key := _interruptor(r)) not in seen:
            seen.add(key)
            new.append(r)
    return Cause((*self.reasons, *new)) if new else self

annotate

annotate(annotations, *, overwrite=False) -> Cause[E_co]

Every reason carrying annotations too (see Fail.annotate).

ParameterTypeDescription
annotationsContext
overwritebool = False

Returns Cause[E_co]

def annotate(self, annotations: Context, *, overwrite: bool = False) -> Cause[E_co]:
    """Every reason carrying ``annotations`` too (see ``Fail.annotate``)."""
    if not annotations.unsafe_map:
        return self
    reasons = tuple(r.annotate(annotations, overwrite=overwrite) for r in self.reasons)
    changed = any(a is not b for a, b in zip(reasons, self.reasons, strict=True))
    return Cause(reasons) if changed else self

map

map(f) -> Cause[E2]

Apply f to every typed error, keeping its annotations; defects and interruptions stay.

ParameterTypeDescription
fCallable[[E_co], E2]

Returns Cause[E2]

def map[E2](self, f: Callable[[E_co], E2]) -> Cause[E2]:
    """Apply ``f`` to every typed error, keeping its annotations;
    defects and interruptions stay."""
    return Cause(
        tuple(
            Fail(f(r.error), annotations=r.annotations) if isinstance(r, Fail) else r
            for r in self.reasons
        )
    )

elect_defects

elect_defects() -> Cause[E_co]

If there is a defect, every typed error becomes a defect too. A Done stays a failure: it is a producer's end, not an error.

Returns Cause[E_co]

def elect_defects(self) -> Cause[E_co]:
    """If there is a defect, every typed error becomes a defect too.
    A ``Done`` stays a failure: it is a producer's end, not an error."""
    if not self.has_dies:
        return self
    from pyfect.errors import Done

    return Cause(
        tuple(
            Die(_as_exception(r.error), annotations=r.annotations)
            if isinstance(r, Fail) and not isinstance(r.error, Done)
            else r
            for r in self.reasons
        )
    )

find_fail

find_fail() -> Fail[E_co] | Cause[Never]

The first Fail, even beside a defect; otherwise this cause, which then holds no typed error. For observing errors (tap_error), never for recovering from them: use find_recoverable_error.

Returns Fail[E_co] | Cause[Never]

def find_fail(self) -> Fail[E_co] | Cause[Never]:
    """The first ``Fail``, even beside a defect; otherwise this cause,
    which then holds no typed error. For observing errors (``tap_error``),
    never for recovering from them: use ``find_recoverable_error``."""
    for r in self.reasons:
        if isinstance(r, Fail):
            return r
    return self.without_failures()

find_error

find_error() -> Result[E_co, Cause[Never]]

Ok of the first typed error, else Err of this cause, which then holds no typed error.

Returns Result[E_co, Cause[Never]]

def find_error(self) -> Result[E_co, Cause[Never]]:
    """``Ok`` of the first typed error, else ``Err`` of this cause, which
    then holds no typed error."""
    found = self.find_fail()
    return Ok(found.error) if isinstance(found, Fail) else Err(found)

find_die

find_die() -> Die | None

Returns Die | None

def find_die(self) -> Die | None:
    return next((r for r in self.reasons if isinstance(r, Die)), None)

find_defect

find_defect() -> BaseException | None

Returns BaseException | None

def find_defect(self) -> BaseException | None:
    die = self.find_die()
    return None if die is None else die.defect

find_interrupt

find_interrupt() -> Interrupt | None

Returns Interrupt | None

def find_interrupt(self) -> Interrupt | None:
    return next((r for r in self.reasons if isinstance(r, Interrupt)), None)

filter_interruptors

filter_interruptors() -> frozenset[int] | None

interruptors if the cause holds an interruption (the set is empty when every one was external), else None.

Returns frozenset[int] | None

def filter_interruptors(self) -> frozenset[int] | None:
    """``interruptors`` if the cause holds an interruption (the set is
    empty when every one was external), else ``None``."""
    return self.interruptors if self.has_interrupts else None

find_recoverable_error

find_recoverable_error() -> Fail[E_co] | Cause[Never]

What a typed recovery handler may act on.

The first Fail, unless the cause holds a defect: defects take precedence, so the result is then the cause to re-fail with, every typed error elected to a defect. A cause with no typed error is returned as is. Interruptions do not block recovery.

Returns Fail[E_co] | Cause[Never]

def find_recoverable_error(self) -> Fail[E_co] | Cause[Never]:
    """What a typed recovery handler may act on.

    The first ``Fail``, unless the cause holds a defect: defects take
    precedence, so the result is then the cause to re-fail with, every
    typed error elected to a defect. A cause with no typed error is
    returned as is. Interruptions do not block recovery.
    """
    if not self.has_dies:
        for r in self.reasons:
            if isinstance(r, Fail):
                return r
        return self.without_failures()
    # Only a ``Done`` is left as a failure, and it is not recoverable.
    elected: Cause[Any] = self.elect_defects()
    return elected

without_failures

without_failures() -> Cause[Never]

Keep defects and interruptions; drop typed errors.

Returns Cause[Never]

def without_failures(self) -> Cause[Never]:
    """Keep defects and interruptions; drop typed errors."""
    return Cause(tuple(r for r in self.reasons if not isinstance(r, Fail)))

squash

squash() -> BaseException

The single exception that best represents this cause.

Priority: the first typed error, then the first defect, then interruption. Typed errors that are not exceptions are wrapped in UnhandledError so they can still be raised.

Returns BaseException

def squash(self) -> BaseException:
    """The single exception that best represents this cause.

    Priority: the first typed error, then the first defect, then
    interruption. Typed errors that are not exceptions are wrapped in
    ``UnhandledError`` so they can still be raised.
    """
    if self.failures:
        return _as_exception(self.failures[0])
    if self.defects:
        return self.defects[0]
    return _interrupted(self) if self.reasons else EmptyCauseError("Empty cause")

pretty_errors

pretty_errors() -> tuple[BaseException, ...]

One exception per typed error and defect, in order. Typed errors that are not exceptions are wrapped in UnhandledError. A cause of interruptions alone gives one Interrupted naming the interruptors; an empty cause gives none.

Returns tuple[BaseException, ...]

def pretty_errors(self) -> tuple[BaseException, ...]:
    """One exception per typed error and defect, in order. Typed errors
    that are not exceptions are wrapped in ``UnhandledError``. A cause of
    interruptions alone gives one ``Interrupted`` naming the interruptors;
    an empty cause gives none."""
    errors = tuple(
        _as_exception(r.error) if isinstance(r, Fail) else r.defect
        for r in self.reasons
        if not isinstance(r, Interrupt)
    )
    return errors if errors or not self.reasons else (_interrupted(self),)

pretty

pretty() -> str

pretty_errors rendered with their tracebacks and cause chains, each followed by the span frames its reason records (Effect's at span (location) lines).

Returns str

def pretty(self) -> str:
    """``pretty_errors`` rendered with their tracebacks and cause chains,
    each followed by the span frames its reason records (Effect's
    ``    at span (location)`` lines)."""
    from pyfect._stack_frame import StackTrace, render

    errors = self.pretty_errors()
    reasons = [r for r in self.reasons if not isinstance(r, Interrupt)] or self.reasons[:1]
    return "\n\n".join(
        "\n".join(
            [
                "".join(traceback.format_exception(e)).rstrip(),
                *render(_frame(r.annotations, StackTrace)),
            ]
        )
        for e, r in zip(errors, reasons, strict=True)
    )