Attributes#
| Attribute | Type | Description |
|---|---|---|
reasons | tuple[Reason[E_co], ...] | |
annotations | Context | Every reason's annotations; a later reason wins a shared key. |
failures | tuple[E_co, ...] | |
defects | tuple[BaseException, ...] | |
interruptors | frozenset[int] | The ids of the fibers that interrupted; external interrupts have none. |
has_fails | bool | |
has_dies | bool | |
has_interrupts | bool | |
has_interrupts_only | bool |
Functions#
__init__#
__init__(reasons) -> None| Parameter | Type | Description |
|---|---|---|
reasons | tuple[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.
| Parameter | Type | Description |
|---|---|---|
interrupts | Cause[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 selfannotate#
annotate(annotations, *, overwrite=False) -> Cause[E_co]Every reason carrying annotations too (see Fail.annotate).
| Parameter | Type | Description |
|---|---|---|
annotations | Context | |
overwrite | bool = 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 selfmap#
map(f) -> Cause[E2]Apply f to every typed error, keeping its annotations;
defects and interruptions stay.
| Parameter | Type | Description |
|---|---|---|
f | Callable[[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 | NoneReturns 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 | NoneReturns BaseException | None
def find_defect(self) -> BaseException | None:
die = self.find_die()
return None if die is None else die.defectfind_interrupt#
find_interrupt() -> Interrupt | NoneReturns 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] | Noneinterruptors 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 Nonefind_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 electedwithout_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() -> BaseExceptionThe 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() -> strpretty_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)
)