GitHub

Typed errors

PreviousNext

Expected failures in the type, handled with catch_tag; defects kept apart.

pyfect separates two kinds of failure, as Effect does:

  • Failures are expected: a user is not found, a request times out. They are values of the error type E, and the type tells callers which ones to handle.
  • Defects are bugs: a ZeroDivisionError, a broken invariant. They are not in the type and ordinary handlers don't see them.

Defining and failing

Subclass fx.TaggedError, usually as a dataclass. Inside a generator, yielding an error fails the effect: return (yield from NotFound(key)). Outside one, use fx.fail(error).

tagged.py
from dataclasses import dataclass

import pyfect as fx


@dataclass
class NotFound(fx.TaggedError):
    key: str


@dataclass
class Forbidden(fx.TaggedError):
    user: str


@fx.fn
def read(user: str, key: str):
    if user != "admin":
        return (yield from Forbidden(user))  # yielding an error fails the effect
    if key != "motd":
        return (yield from NotFound(key))
    return "hello"


# read("admin", "motd") is an Effect[str, Forbidden | NotFound].
exit = fx.run_sync_exit(read("guest", "motd"))
assert exit == fx.exit.fail(Forbidden("guest"))

Handling errors

catch_tag(ErrorClass, handler) handles one error class and removes it from the type. Because it removes something, it is a function you use with .pipe(...), one per call; pyright cannot infer the removal from a method. catch_all handles every typed error.

Other handlers: catch_if, map_error, or_else_succeed, match, result (an Ok/Err value), ignore, and retry (see Schedules and retries).

catch_tag.py
from dataclasses import dataclass

import pyfect as fx


@dataclass
class NotFound(fx.TaggedError):
    key: str


@dataclass
class Forbidden(fx.TaggedError):
    user: str


@fx.fn
def read(user: str, key: str):
    if user != "admin":
        return (yield from Forbidden(user))
    if key != "motd":
        return (yield from NotFound(key))
    return "hello"


# catch_tag removes exactly one error from the type, so it goes through .pipe.
# Effect[str, Forbidden]: NotFound is handled, Forbidden is left.
with_default = read("admin", "news").pipe(fx.catch_tag(NotFound, lambda e: fx.succeed("")))

# catch_all handles every typed error: Effect[str] (no errors left).
safe = with_default.catch_all(lambda e: fx.succeed(f"denied: {e.user}"))

assert fx.run_sync(safe) == ""
assert fx.run_sync(read("guest", "motd").catch_all(lambda e: fx.succeed(repr(e)))) == (
    "Forbidden(user='guest')"
)

Defects

An exception raised inside fx.sync or a map is a defect. Use fx.try_ (or try_async, try_blocking) with catch= to turn an exception you expect into a typed failure.

When a failure and a defect happen together (a finalizer crashed while handling an error), the defect wins and typed handlers don't run, so a bug is never mistaken for a handled error. A failed run's Exit holds a Cause; cause.pretty() prints every failure, defect and interruption in it with tracebacks.

defects.py
from dataclasses import dataclass

import pyfect as fx


@dataclass
class ParseError(fx.TaggedError):
    text: str


def parse(text: str) -> fx.Effect[int, ParseError]:
    # try_ turns an exception into a typed failure with catch=.
    return fx.try_(lambda: int(text), catch=lambda _: ParseError(text))


assert fx.run_sync_exit(parse("42")) == fx.Success(42)

# A defect is a bug, not an expected failure: it is not in the error type,
# and catch_all does not see it.
buggy = fx.sync(lambda: 1 // 0).catch_all(lambda _: fx.succeed(0))
exit = fx.run_sync_exit(buggy)
assert isinstance(exit, fx.Failure)
assert isinstance(exit.cause.squash(), ZeroDivisionError)

# catch_defect is there for the edges of a program (logging, a 500 response).
recovered = fx.sync(lambda: 1 // 0).catch_defect(lambda _: fx.succeed(-1))
assert fx.run_sync(recovered) == -1

# or_die turns typed errors into defects when they can't happen.
assert fx.run_sync(parse("7").or_die()) == 7

Reference: pyfect.errors, pyfect.cause, pyfect.exit.