Config

__init__(evaluate) -> None

def __init__(self, evaluate: Evaluate[A_co]) -> None:
    self._evaluate = evaluate

parse(provider) -> Effect[A_co, ConfigError]

def parse(self, provider: ConfigProvider) -> Effect[A_co, ConfigError]:
    """Read this config from ``provider``; absence fails too."""
    return self._evaluate(provider, ()).flat_map(_present_or_fail)

map(f) -> Config[B]

@override
def map[B](self, f: Callable[[A_co], B]) -> Config[B]:
    evaluate = self._evaluate
    return Config(lambda provider, prefix: evaluate(provider, prefix).map(lambda r: r.map(f)))

flat_map(f) -> Effect[Any, Any, Any]

@override
def flat_map(self, f: Callable[[A_co], Effect[Any, Any, Any]]) -> Effect[Any, Any, Any]:
    """Choose the next config from this one's value; it is read under
    the same ``nested`` prefix. Absence of either is absence. A function
    returning a plain effect gives a plain effect, as ``Effect.flat_map``
    does."""
    evaluate = self._evaluate

    def run(provider: ConfigProvider, prefix: Path) -> Effect[Resolution[Any], Any]:
        def next_config(resolution: Resolution[A_co]) -> Effect[Resolution[Any], Any]:
            match resolution:
                case Ok(value):
                    return _evaluate_at(f(value), provider, prefix)
                case Err():
                    return succeed(resolution)

        return evaluate(provider, prefix).flat_map(next_config)

    return Config(run)

map_effect(f) -> Config[B]

def map_effect[B](self, f: Callable[[A_co], Effect[B, ConfigError]]) -> Config[B]:
    """Transform the value with an effect that may fail (Effect's
    ``mapEffect``); its failure is a failure, not absence."""
    evaluate = self._evaluate

    def next_value(resolution: Resolution[A_co]) -> Effect[Resolution[B], ConfigError]:
        match resolution:
            case Ok(value):
                return f(value).map(Ok)
            case Err():
                return succeed(resolution)

    return Config(lambda provider, prefix: evaluate(provider, prefix).flat_map(next_value))

or_else(that) -> Config[A_co | B]

def or_else[B](self, that: Callable[[ConfigError], Config[B]]) -> Config[A_co | B]:
    """Read ``that(error)`` when this config is absent or fails with a
    ``ConfigError``. Defects and interruptions are not recovered, nor is
    a failure that comes with one."""
    evaluate = self._evaluate

    def run(
        provider: ConfigProvider, prefix: Path
    ) -> Effect[Resolution[A_co | B], ConfigError]:
        def on_failure(cause: Cause[ConfigError]) -> Effect[Resolution[B], ConfigError]:
            match cause.reasons:
                case (Fail(error), *rest) if all(isinstance(r, Fail) for r in rest):
                    return that(error)._evaluate(provider, prefix)
                case _:
                    return fail_cause(cause)

        def on_success(
            resolution: Resolution[A_co],
        ) -> Effect[Resolution[A_co | B], ConfigError]:
            match resolution:
                case Ok():
                    return succeed(resolution)
                case Err(error):
                    return that(error)._evaluate(provider, prefix)

        return evaluate(provider, prefix).match_cause_effect(
            on_failure=on_failure, on_success=on_success
        )

    return Config(run)

with_default(default) -> Config[A_co | B]

def with_default[B](self, default: B) -> Config[A_co | B]:
    """``default`` when the value is absent. Invalid data and source
    errors still fail. A group from ``all`` is absent as a whole when
    any part is."""
    evaluate = self._evaluate
    present: Resolution[B] = Ok(default)

    def recover(resolution: Resolution[A_co]) -> Resolution[A_co | B]:
        return present if isinstance(resolution, Err) else resolution

    return Config(lambda provider, prefix: evaluate(provider, prefix).map(recover))

nested(name) -> Config[A_co]

def nested(self, name: str) -> Config[A_co]:
    """Read this config under ``name``: an object level in
    ``from_unknown``, a ``name_`` prefix in ``from_env``. The outermost
    ``nested`` is the outermost segment."""
    evaluate = self._evaluate
    return Config(lambda provider, prefix: evaluate(provider, (*prefix, name)))

delay(duration) -> Effect[A_co, E_co, R_co]

def delay(self, duration: Duration) -> Effect[A_co, E_co, R_co]:
    from pyfect.clock import sleep

    return sleep(duration).and_then(as_effect(self))

timed() -> Effect[tuple[float, A_co], E_co, R_co]

def timed(self) -> Effect[tuple[float, A_co], E_co, R_co]:
    """Pair the value with how long it took, in seconds."""
    from pyfect.clock import timed

    return timed(as_effect(self))

cached() -> Effect[Effect[A_co, E_co, R_co]]

def cached(self) -> Effect[Effect[A_co, E_co, R_co]]:
    """An effect that runs this one on first use and then replays its
    exit. Concurrent callers share one run; an interrupted run is not
    cached (see ``cached_invalidate_with_ttl``)."""
    from pyfect._cached import cached

    return cached(as_effect(self))

cached_with_ttl(time_to_live) -> Effect[Effect[A_co, E_co, R_co]]

def cached_with_ttl(
    self, time_to_live: Duration | Callable[[Exit[A_co, E_co]], Duration]
) -> Effect[Effect[A_co, E_co, R_co]]:
    """``cached``, but each exit is replayed only for ``time_to_live``
    (or ``time_to_live(exit)``) after it was produced."""
    from pyfect._cached import cached_invalidate_with_ttl

    return cached_invalidate_with_ttl(as_effect(self), time_to_live).map(lambda pair: pair[0])

cached_invalidate_with_ttl(time_to_live) -> Effect[tuple[Effect[A_co, E_co, R_co], Effect[None]]]

def cached_invalidate_with_ttl(
    self, time_to_live: Duration | Callable[[Exit[A_co, E_co]], Duration]
) -> Effect[tuple[Effect[A_co, E_co, R_co], Effect[None]]]:
    """``cached_with_ttl`` plus an effect that drops the cached exit.

    Callers already waiting on a run keep waiting on it; later callers
    start a fresh run. A run is interrupted once every caller waiting on
    it has gone, and an interrupted run is never cached."""
    from pyfect._cached import cached_invalidate_with_ttl

    return cached_invalidate_with_ttl(as_effect(self), time_to_live)

retry(policy=None, /, *, schedule=None, times=None, while_=None, until=None) -> Effect[Any, Any, Any]

eff.retry(lambda typed: typed(schedule.recurs(3)).while_(lambda m: m.input.retryable))

def retry(
    self,
    policy: Schedule[Any, Any, Any, Any] | Callable[[Any], Any] | None = None,
    /,
    *,
    schedule: Schedule[Any, Any, Any, Any] | None = None,
    times: int | None = None,
    while_: Callable[[Any], Any] | None = None,
    until: Callable[[Any], Any] | None = None,
) -> Effect[Any, Any, Any]:
    """Retry on typed failure while ``policy`` recurs; then fail with the
    last error. Defects and interruptions are not retried.

    ``policy`` is a schedule, or a function from ``SetInputType`` to
    one, so the schedule's metadata sees this effect's error type::

        eff.retry(lambda typed: typed(schedule.recurs(3)).while_(lambda m: m.input.retryable))

    Instead of a policy, the keywords build one: retry at most
    ``times`` times, only while ``while_(error)`` holds, until
    ``until(error)`` holds, on ``schedule``'s delays. Predicates may
    return a bool or an effect of one; with a ``TypeIs`` predicate as
    ``until`` (and no ``schedule`` or ``times``), the error narrows."""
    from pyfect.schedule._drive import resolve_policy, retry

    policy = resolve_policy(policy, schedule, times, while_, until)
    return retry(as_effect(self), policy)

retry_or_else(policy, or_else) -> Effect[A_co | A2, EE, R_co | RR]

def retry_or_else[B, A2, EE = Never, RR = Never](
    self,
    policy: Schedule[B, E_co, EE, RR],
    or_else: Callable[[E_co, B], Effect[A2, EE, RR]],
) -> Effect[A_co | A2, EE, R_co | RR]:
    """``retry``, handing the last error and the policy's final output
    to ``or_else`` once the policy stops."""
    from pyfect.schedule._drive import retry_or_else

    return retry_or_else(as_effect(self), policy, or_else)

repeat(policy=None, /, *, schedule=None, times=None, while_=None, until=None) -> Effect[Any, Any, Any]

def repeat(
    self,
    policy: Schedule[Any, Any, Any, Any] | Callable[[Any], Any] | None = None,
    /,
    *,
    schedule: Schedule[Any, Any, Any, Any] | None = None,
    times: int | None = None,
    while_: Callable[[Any], Any] | None = None,
    until: Callable[[Any], Any] | None = None,
) -> Effect[Any, Any, Any]:
    """Repeat after each success while the schedule recurs. Succeeds
    with the schedule's final output. The schedule may also be given
    as a function from ``SetInputType`` to one (see ``retry``).

    Instead of a schedule, the keywords build one: repeat at most
    ``times`` times, only while ``while_(value)`` holds, until
    ``until(value)`` holds, on ``schedule``'s delays. That form
    succeeds with the last value, narrowed by a ``TypeIs`` predicate
    as ``until`` (with no ``schedule`` or ``times``). Predicates may
    return a bool or an effect of one."""
    from pyfect.schedule._drive import repeat, resolve_policy

    policy = resolve_policy(policy, schedule, times, while_, until)
    return repeat(as_effect(self), policy)

repeat_or_else(schedule, or_else) -> Effect[B | B2, E2, R_co | RR]

def repeat_or_else[B, B2, E2, EE = Never, RR = Never](
    self,
    schedule: Schedule[B, A_co, EE, RR],
    or_else: Callable[[E_co | EE, Metadata[B, A_co] | None], Effect[B2, E2, RR]],
) -> Effect[B | B2, E2, R_co | RR]:
    """``repeat``, handing a typed failure (of this effect or of the
    schedule) and the last step's metadata (``None`` before the first)
    to ``or_else``."""
    from pyfect.schedule._drive import repeat_or_else

    return repeat_or_else(as_effect(self), schedule, or_else)

schedule(schedule) -> Effect[B, E_co | EE, R_co | RR]

def schedule[B, EE = Never, RR = Never](
    self, schedule: Schedule[B, A_co | None, EE, RR]
) -> Effect[B, E_co | EE, R_co | RR]:
    """Run after each recurrence of ``schedule`` (first after its first
    delay), stepping it with ``None`` and then each result. Succeeds
    with the schedule's final output."""
    from pyfect.schedule._drive import scheduled

    return scheduled(as_effect(self), schedule)

schedule_from(initial, schedule) -> Effect[B, E_co | EE, R_co | RR]

def schedule_from[A, B, EE = Never, RR = Never](
    self: TimingMethods[A, E_co, R_co], initial: A, schedule: Schedule[B, A, EE, RR]
) -> Effect[B, E_co | EE, R_co | RR]:
    """``schedule``, stepping first with ``initial``."""
    from pyfect.schedule._drive import schedule_from

    return schedule_from(as_effect(self), initial, schedule)

fork(*, start_immediately=False, uninterruptible=False) -> Effect[Fiber[A_co, E_co], Never, R_co]

def fork(
    self, *, start_immediately: bool = False, uninterruptible: Uninterruptible = False
) -> Effect[Fiber[A_co, E_co], Never, R_co]:
    """Start this effect on a child fiber. The child is interrupted if
    it is still running when the parent fiber finishes.

    With ``start_immediately`` the child runs at once, up to its first
    wait, before the parent goes on; otherwise it starts on the next
    scheduler turn. ``uninterruptible`` starts it uninterruptible
    (``True``) or as interruptible as the parent is (``"inherit"``)."""
    return _core._Fork(
        as_effect(self),
        daemon=False,
        start_immediately=start_immediately,
        uninterruptible=uninterruptible,
    )

fork_daemon(*, start_immediately=False, uninterruptible=False) -> Effect[Fiber[A_co, E_co], Never, R_co]

def fork_daemon(
    self, *, start_immediately: bool = False, uninterruptible: Uninterruptible = False
) -> Effect[Fiber[A_co, E_co], Never, R_co]:
    """Start this effect on a fiber that outlives its parent. Options as
    for ``fork``."""
    return _core._Fork(
        as_effect(self),
        daemon=True,
        start_immediately=start_immediately,
        uninterruptible=uninterruptible,
    )

fork_scoped(*, start_immediately=False, uninterruptible=False) -> Effect[Fiber[A_co, E_co], Never, R_co | Scope]

def fork_scoped(
    self, *, start_immediately: bool = False, uninterruptible: Uninterruptible = False
) -> Effect[Fiber[A_co, E_co], Never, R_co | Scope]:
    """``fork_in`` the enclosing ``Scope``."""
    from pyfect.scope import fork_scoped

    return fork_scoped(
        as_effect(self), start_immediately=start_immediately, uninterruptible=uninterruptible
    )

fork_in(scope, *, start_immediately=False, uninterruptible=False) -> Effect[Fiber[A_co, E_co], Never, R_co]

def fork_in(
    self,
    scope: CloseableScope,
    *,
    start_immediately: bool = False,
    uninterruptible: Uninterruptible = False,
) -> Effect[Fiber[A_co, E_co], Never, R_co]:
    """Start this effect on a fiber that outlives its parent and is
    interrupted when ``scope`` closes. Into a closed scope it gives an
    interrupted fiber that never ran. Options as for ``fork``."""
    from pyfect.scope import fork_in

    return fork_in(
        as_effect(self),
        scope,
        start_immediately=start_immediately,
        uninterruptible=uninterruptible,
    )

await_all_children() -> Effect[A_co, E_co, R_co]

def await_all_children(self) -> Effect[A_co, E_co, R_co]:
    """When this effect ends, wait for the child fibers it forked
    (not those forked before it started). The wait is interruptible
    if the fiber was interruptible when this effect started."""
    from pyfect._fiber_all import await_all_children

    return await_all_children(as_effect(self))

race(that) -> Effect[A_co | B, E_co | E2, R_co | R2]

def race[B, E2 = Never, R2 = Never](
    self, that: Effect[B, E2, R2]
) -> Effect[A_co | B, E_co | E2, R_co | R2]:
    """The first to succeed wins and the other is interrupted. Fails
    only if both fail."""
    require_effect(that, "that")
    from pyfect._concurrency import race

    return race(as_effect(self), that)

race_first(that) -> Effect[A_co | B, E_co | E2, R_co | R2]

def race_first[B, E2 = Never, R2 = Never](
    self, that: Effect[B, E2, R2]
) -> Effect[A_co | B, E_co | E2, R_co | R2]:
    """The first to finish wins, success or failure."""
    require_effect(that, "that")
    from pyfect._concurrency import race_first

    return race_first(as_effect(self), that)

timeout(duration) -> Effect[A_co, E_co | TimeoutException, R_co]

def timeout(self, duration: Duration) -> Effect[A_co, E_co | TimeoutException, R_co]:
    """Fail with ``TimeoutException`` if this takes longer than
    ``duration``. The effect is interrupted when time runs out."""
    from pyfect._concurrency import timeout

    return timeout(as_effect(self), duration)

timeout_option(duration) -> Effect[Option[A_co], E_co, R_co]

def timeout_option(self, duration: Duration) -> Effect[Option[A_co], E_co, R_co]:
    """``Some(value)``, or ``None`` if this takes longer than
    ``duration``. The effect is interrupted when time runs out."""
    from pyfect._concurrency import timeout_option

    return timeout_option(as_effect(self), duration)

timeout_or_else(duration, or_else) -> Effect[A_co | B, E_co | E2, R_co | R2]

def timeout_or_else[B, E2 = Never, R2 = Never](
    self, duration: Duration, or_else: Callable[[], Effect[B, E2, R2]]
) -> Effect[A_co | B, E_co | E2, R_co | R2]:
    """Continue with ``or_else()`` if this takes longer than
    ``duration``. ``or_else`` runs once this effect has been
    interrupted and has finished cleaning up."""
    from pyfect._concurrency import timeout_or_else

    return timeout_or_else(as_effect(self), duration, or_else)

track(metric, f=None) -> Effect[A_co, E_co, R_co]

def track(
    self, metric: Metric[Any, object], f: Callable[[Any], Any] | None = None, /
) -> Effect[A_co, E_co, R_co]:
    """Update ``metric`` with this effect's exit (or ``f(exit)``)
    however it ends."""
    return as_effect(self).on_exit(lambda ex: metric.update(ex if f is None else f(ex)))

track_successes(metric, f=None) -> Effect[A_co, E_co, R_co]

def track_successes(
    self, metric: Metric[Any, object], f: Callable[[Any], Any] | None = None, /
) -> Effect[A_co, E_co, R_co]:
    """Update ``metric`` with each value (or ``f(value)``)."""
    return as_effect(self).tap(lambda a: metric.update(a if f is None else f(a)))

track_errors(metric, f=None) -> Effect[A_co, E_co, R_co]

def track_errors(
    self, metric: Metric[Any, object], f: Callable[[Any], Any] | None = None, /
) -> Effect[A_co, E_co, R_co]:
    """Update ``metric`` with each typed error (or ``f(error)``)."""
    return as_effect(self).tap_error(lambda e: metric.update(e if f is None else f(e)))

track_defects(metric, f=None) -> Effect[A_co, E_co, R_co]

def track_defects(
    self, metric: Metric[Any, object], f: Callable[[Any], Any] | None = None, /
) -> Effect[A_co, E_co, R_co]:
    """Update ``metric`` with each defect (or ``f(defect)``)."""
    return as_effect(self).tap_defect(lambda d: metric.update(d if f is None else f(d)))

track_duration(metric, f=None) -> Effect[A_co, E_co, R_co]

def track_duration(
    self, metric: Metric[Any, object], f: Callable[[Any], Any] | None = None, /
) -> Effect[A_co, E_co, R_co]:
    """Update ``metric`` with how long this effect took (or
    ``f(duration)``), measured on the ``Clock``'s monotonic time
    however it ends."""
    from pyfect.clock import Clock
    from pyfect.duration import Seconds

    inner = as_effect(self)

    def body(fiber: Fiber[Any, Any]) -> Effect[A_co, E_co, R_co]:
        clock = fiber.get_ref(Clock)
        start = clock.unsafe_monotonic_nanos()

        def record(_: object) -> Effect[None]:
            elapsed = Seconds((clock.unsafe_monotonic_nanos() - start) / 1_000_000_000)
            return metric.update(elapsed if f is None else f(elapsed))

        return inner.on_exit(record)

    return _core._WithFiber(body)

with_span(name, **options) -> Effect[A_co, E_co, R_co]

def with_span(
    self, name: str, **options: Unpack[TracedSpanOptions]
) -> Effect[A_co, E_co, R_co]:
    """Run inside a new span, the parent of the spans made inside. It
    ends with this effect's exit. Failures inside record the span's
    stack frame: this call site unless ``capture_stack_trace`` says
    otherwise."""
    from pyfect._stack_frame import stack_of
    from pyfect.tracer._spans import in_span_frame, span_using

    eff = as_effect(self)
    stack = stack_of(options.get("capture_stack_trace", True), 1)
    return span_using(name, lambda span: in_span_frame(eff, span, stack), options)

with_span_scoped(name, **options) -> Effect[A_co, E_co, R_co | Scope]

def with_span_scoped(
    self, name: str, **options: Unpack[TracedSpanOptions]
) -> Effect[A_co, E_co, R_co | Scope]:
    """Run inside a new span that ends when the current scope closes."""
    from pyfect._stack_frame import stack_of
    from pyfect.tracer._spans import in_span_frame, span_scoped

    eff = as_effect(self)
    stack = stack_of(options.get("capture_stack_trace", True), 1)
    return span_scoped(name, options).flat_map(lambda span: in_span_frame(eff, span, stack))

with_parent_span(span, *, capture_stack_trace=True) -> Effect[A_co, E_co, R_co]

def with_parent_span(
    self, span: AnySpan, *, capture_stack_trace: CaptureStackTrace = True
) -> Effect[A_co, E_co, R_co]:
    """Run with ``span`` as the parent of the spans made inside, and,
    unless it is external, inside its stack frame."""
    from pyfect._stack_frame import stack_of
    from pyfect.tracer._spans import in_span_frame

    return in_span_frame(as_effect(self), span, stack_of(capture_stack_trace, 1))

annotate_spans(key_or_values, value=None) -> Effect[A_co, E_co, R_co]

def annotate_spans(
    self, key_or_values: str | Mapping[str, object], value: object = None, /
) -> Effect[A_co, E_co, R_co]:
    """Run with these attributes set on every span made inside."""
    from pyfect.tracer._refs import TracerSpanAnnotations

    added = {key_or_values: value} if isinstance(key_or_values, str) else dict(key_or_values)
    return as_effect(self).update_service(
        TracerSpanAnnotations, lambda annotations: {**annotations, **added}
    )

link_spans(*spans, /, *, attributes=None) -> Effect[A_co, E_co, R_co]

def link_spans(
    self, *spans: AnySpan, attributes: Mapping[str, object] | None = None
) -> Effect[A_co, E_co, R_co]:
    """Run with links to ``spans`` added to every span made inside."""
    from pyfect.tracer._model import SpanLink
    from pyfect.tracer._refs import TracerSpanLinks

    shared: Mapping[str, object] = {} if attributes is None else attributes
    links = tuple(SpanLink(span, shared) for span in spans)
    return as_effect(self).update_service(TracerSpanLinks, lambda current: (*current, *links))

with_tracer(tracer) -> Effect[A_co, E_co, R_co]

def with_tracer(self, tracer: Tracer) -> Effect[A_co, E_co, R_co]:
    """Run with spans made by ``tracer``."""
    from pyfect._tracer_ref import CurrentTracer

    return as_effect(self).update_service(CurrentTracer, lambda _: tracer)

with_tracer_enabled(enabled) -> Effect[A_co, E_co, R_co]

def with_tracer_enabled(self, enabled: bool) -> Effect[A_co, E_co, R_co]:
    """Run with tracing on or off: off, every span is a no-op span."""
    from pyfect.tracer._refs import TracerEnabled

    return as_effect(self).update_service(TracerEnabled, lambda _: enabled)

with_tracer_timing(enabled) -> Effect[A_co, E_co, R_co]

def with_tracer_timing(self, enabled: bool) -> Effect[A_co, E_co, R_co]:
    """Run with span timing on or off: off, spans start and end at 0."""
    from pyfect.tracer._refs import TracerTimingEnabled

    return as_effect(self).update_service(TracerTimingEnabled, lambda _: enabled)

annotate_logs(key_or_values, value=None) -> Effect[A_co, E_co, R_co]

def annotate_logs(
    self, key_or_values: str | Mapping[str, object], value: object = None, /
) -> Effect[A_co, E_co, R_co]:
    """Run with these annotations added to every log."""
    from pyfect.logger import CurrentLogAnnotations

    added = {key_or_values: value} if isinstance(key_or_values, str) else dict(key_or_values)
    return as_effect(self).update_service(
        CurrentLogAnnotations, lambda annotations: {**annotations, **added}
    )

with_log_span(label) -> Effect[A_co, E_co, R_co]

def with_log_span(self, label: str) -> Effect[A_co, E_co, R_co]:
    """Run inside a log span: logs show ``label=<elapsed>ms``, timed by
    the ``Clock`` from when this effect starts."""
    from pyfect.clock import Clock
    from pyfect.logger import CurrentLogSpans, LogSpan

    inner = as_effect(self)

    def body(fiber: Fiber[Any, Any]) -> Effect[A_co, E_co, R_co]:
        span = LogSpan(label, fiber.get_ref(Clock).unsafe_now_nanos() // 1_000_000)
        return inner.update_service(CurrentLogSpans, lambda spans: (span, *spans))

    return _core._WithFiber(body)

with_logger(logger) -> Effect[A_co, E_co, R_co]

def with_logger(self, logger: Logger[object]) -> Effect[A_co, E_co, R_co]:
    """Run with ``logger`` added to the current loggers."""
    from pyfect.logger import CurrentLoggers

    return as_effect(self).update_service(
        CurrentLoggers,
        lambda loggers: loggers if any(lg is logger for lg in loggers) else (*loggers, logger),
    )

update_service(tag, f) -> Effect[A_co, E_co, R_co | T]

def update_service[S, T](
    self, tag: TagLike[S, T], f: Callable[[S], S]
) -> Effect[A_co, E_co, R_co | T]:
    """Run with ``f`` applied to the service of ``tag``, which this
    effect then needs."""
    return _core._Locally(as_effect(self), lambda ctx: ctx.add(tag, f(ctx.unsafe_get(tag))))

on_exit(f) -> Effect[A_co, E_co, R_co | R2]

def on_exit[R2 = Never](
    self, f: Callable[[Exit[A_co, E_co]], Effect[object, Never, R2]]
) -> Effect[A_co, E_co, R_co | R2]:
    """Run ``f`` after this effect however it ends. ``f`` runs
    uninterruptibly; if it fails, its cause is added to the result."""
    return _on_exit(as_effect(self), f)

ensuring(finalizer) -> Effect[A_co, E_co, R_co | R2]

def ensuring[R2 = Never](
    self, finalizer: Effect[object, Never, R2]
) -> Effect[A_co, E_co, R_co | R2]:
    require_effect(finalizer, "finalizer")
    return self.on_exit(lambda _: finalizer)

on_error(f) -> Effect[A_co, E_co, R_co | R2]

def on_error[R2 = Never](
    self, f: Callable[[Cause[E_co]], Effect[object, Never, R2]]
) -> Effect[A_co, E_co, R_co | R2]:
    def handle(ex: Exit[A_co, E_co]) -> Effect[object, Never, R2]:
        return f(ex.cause) if isinstance(ex, Failure) else _core._Succeed(None)

    return self.on_exit(handle)

on_error_if(predicate, f) -> Effect[A_co, E_co | E2, R_co | R2]

def on_error_if[E2 = Never, R2 = Never](
    self,
    predicate: Callable[[Cause[E_co]], bool],
    f: Callable[[Cause[E_co]], Effect[object, E2, R2]],
) -> Effect[A_co, E_co | E2, R_co | R2]:
    """``on_error`` for the causes ``predicate`` accepts. If ``f``
    fails, its cause is added to the result."""

    def handle(ex: Exit[A_co, E_co]) -> Effect[object, E2, R2]:
        if isinstance(ex, Failure) and predicate(ex.cause):
            return f(ex.cause)
        return _core._Succeed(None)

    return _on_exit(as_effect(self), handle)

on_error_filter(filter, f) -> Effect[A_co, E_co | E2, R_co | R2]

def on_error_filter[B, E2 = Never, R2 = Never](
    self,
    filter: Callable[[Cause[E_co]], Result[B, object]],
    f: Callable[[B, Cause[E_co]], Effect[object, E2, R2]],
) -> Effect[A_co, E_co | E2, R_co | R2]:
    """``on_error`` for the causes ``filter`` passes (``Ok(value)``):
    run ``f(value, cause)``."""

    def handle(ex: Exit[A_co, E_co]) -> Effect[object, E2, R2]:
        if isinstance(ex, Failure):
            passed = filter(ex.cause)
            if isinstance(passed, Ok):
                return f(passed.value, ex.cause)
        return _core._Succeed(None)

    return _on_exit(as_effect(self), handle)

on_exit_if(predicate, f) -> Effect[A_co, E_co | E2, R_co | R2]

def on_exit_if[E2 = Never, R2 = Never](
    self,
    predicate: Callable[[Exit[A_co, E_co]], bool],
    f: Callable[[Exit[A_co, E_co]], Effect[object, E2, R2]],
) -> Effect[A_co, E_co | E2, R_co | R2]:
    """``on_exit`` for the exits ``predicate`` accepts. If ``f`` fails,
    its cause is added to the result."""

    def handle(ex: Exit[A_co, E_co]) -> Effect[object, E2, R2]:
        return f(ex) if predicate(ex) else _core._Succeed(None)

    return _on_exit(as_effect(self), handle)

on_exit_filter(filter, f) -> Effect[A_co, E_co | E2, R_co | R2]

def on_exit_filter[B, E2 = Never, R2 = Never](
    self,
    filter: Callable[[Exit[A_co, E_co]], Result[B, object]],
    f: Callable[[B, Exit[A_co, E_co]], Effect[object, E2, R2]],
) -> Effect[A_co, E_co | E2, R_co | R2]:
    """``on_exit`` for the exits ``filter`` passes (``Ok(value)``): run
    ``f(value, exit)``."""

    def handle(ex: Exit[A_co, E_co]) -> Effect[object, E2, R2]:
        passed = filter(ex)
        return f(passed.value, ex) if isinstance(passed, Ok) else _core._Succeed(None)

    return _on_exit(as_effect(self), handle)

on_interrupt(finalizer) -> Effect[A_co, E_co, R_co | R2]

def on_interrupt[R2 = Never](
    self, finalizer: Effect[object, Never, R2]
) -> Effect[A_co, E_co, R_co | R2]:
    require_effect(finalizer, "finalizer")

    def handle(ex: Exit[A_co, E_co]) -> Effect[object, Never, R2]:
        if isinstance(ex, Failure) and ex.cause.has_interrupts:
            return finalizer
        return _core._Succeed(None)

    return self.on_exit(handle)

uninterruptible() -> Effect[A_co, E_co, R_co]

def uninterruptible(self) -> Effect[A_co, E_co, R_co]:
    return _core._SetInterruptible(as_effect(self), False)

interruptible() -> Effect[A_co, E_co, R_co]

def interruptible(self) -> Effect[A_co, E_co, R_co]:
    return _core._SetInterruptible(as_effect(self), True)

match(*, on_failure, on_success) -> Effect[B | C, Never, R_co]

def match[B, C](
    self, *, on_failure: Callable[[E_co], B], on_success: Callable[[A_co], C]
) -> Effect[B | C, Never, R_co]:
    return _core._MatchCause(
        as_effect(self),
        _recovering(lambda e: _core._Succeed(on_failure(e))),
        lambda a: _core._Succeed(on_success(a)),
    )

match_cause(*, on_failure, on_success) -> Effect[B | C, Never, R_co]

def match_cause[B, C](
    self, *, on_failure: Callable[[Cause[E_co]], B], on_success: Callable[[A_co], C]
) -> Effect[B | C, Never, R_co]:
    """``match`` over the whole ``Cause``: defects reach ``on_failure``
    too. Interruption still propagates."""
    return _core._MatchCause(
        as_effect(self),
        lambda c: _core._Succeed(on_failure(c)),
        lambda a: _core._Succeed(on_success(a)),
    )

option() -> Effect[Option[A_co], Never, R_co]

def option(self) -> Effect[Option[A_co], Never, R_co]:
    """Succeed with ``Some(value)``, or ``None`` on a typed error.
    Defects and interruptions still propagate."""
    from pyfect.option import Some

    return _core._MatchCause(
        as_effect(self), _recovering(_succeed_none), lambda a: _core._Succeed(Some(a))
    )

match_effect(*, on_failure, on_success) -> Effect[B | C, EE, R_co | RR]

def match_effect[B, C, EE = Never, RR = Never](
    self,
    *,
    on_failure: Callable[[E_co], Effect[B, EE, RR]],
    on_success: Callable[[A_co], Effect[C, EE, RR]],
) -> Effect[B | C, EE, R_co | RR]:
    return _core._MatchCause(as_effect(self), _recovering(on_failure), on_success)

match_cause_effect(*, on_failure, on_success) -> Effect[B | C, EE, R_co | RR]

def match_cause_effect[B, C, EE = Never, RR = Never](
    self,
    *,
    on_failure: Callable[[Cause[E_co]], Effect[B, EE, RR]],
    on_success: Callable[[A_co], Effect[C, EE, RR]],
) -> Effect[B | C, EE, R_co | RR]:
    return _core._MatchCause(as_effect(self), on_failure, on_success)

exit() -> Effect[Exit[A_co, E_co], Never, R_co]

def exit(self) -> Effect[Exit[A_co, E_co], Never, R_co]:
    """Succeed with the outcome as a value. Interruption still
    propagates; it is not something a fiber can catch."""
    return _core._MatchCause(
        as_effect(self),
        lambda c: _core._Succeed(Failure(c)),
        lambda a: _core._Succeed(Success(a)),
    )

result() -> Effect[Ok[A_co] | Err[E_co], Never, R_co]

def result(self) -> Effect[Ok[A_co] | Err[E_co], Never, R_co]:
    """Succeed with ``Ok(value)`` or ``Err(error)``. Defects and
    interruptions still propagate."""
    return _core._MatchCause(
        as_effect(self),
        _recovering(lambda e: _core._Succeed(Err(e))),
        lambda a: _core._Succeed(Ok(a)),
    )

map_error(f) -> Effect[A_co, E2, R_co]

def map_error[E2](self, f: Callable[[E_co], E2]) -> Effect[A_co, E2, R_co]:
    def on_failure(cause: Cause[E_co]) -> Effect[Never, E2, Never]:
        return _core._FailCause(cause.map(f))

    return _core._MatchCause(as_effect(self), on_failure, None)

map_both(*, on_failure, on_success) -> Effect[B, E2, R_co]

def map_both[B, E2](
    self, *, on_failure: Callable[[E_co], E2], on_success: Callable[[A_co], B]
) -> Effect[B, E2, R_co]:
    """``map_error`` and ``map`` at once."""
    return _core._MatchCause(
        as_effect(self),
        lambda c: _core._FailCause(c.map(on_failure)),
        lambda a: Success(on_success(a)),
    )

catch_all(f) -> Effect[A_co | B, E2, R_co | R2]

def catch_all[B, E2 = Never, R2 = Never](
    self, f: Callable[[E_co], Effect[B, E2, R2]]
) -> Effect[A_co | B, E2, R_co | R2]:
    """Recover from every typed error. Defects and interruptions pass
    through. To handle one error class, use ``pipe(catch_tag(...))``."""
    return _core._MatchCause(as_effect(self), _recovering(f), None)

catch_if(predicate, f, *, or_else=None) -> Effect[Any, Any, Any]

def catch_if(
    self,
    predicate: Callable[[Any], bool],
    f: Callable[[Any], Effect[Any, Any, Any]],
    *,
    or_else: Callable[[Any], Effect[Any, Any, Any]] | None = None,
) -> Effect[Any, Any, Any]:
    """Recover from the typed errors that pass ``predicate``; ``or_else``,
    if given, handles the rest. To remove the caught errors from ``E``,
    use ``pipe(catch_if(refinement, ...))`` with a ``TypeIs`` refinement."""
    return _core._MatchCause(as_effect(self), _catching_if(predicate, f, or_else), None)

catch_filter(filter, f, *, or_else=None) -> Effect[Any, Any, Any]

def catch_filter(
    self,
    filter: Callable[[Any], Result[Any, Any]],
    f: Callable[[Any], Effect[Any, Any, Any]],
    *,
    or_else: Callable[[Any], Effect[Any, Any, Any]] | None = None,
) -> Effect[Any, Any, Any]:
    """Recover with ``f(value)`` from a typed error ``filter`` maps to
    ``Ok(value)``. One it maps to ``Err(rest)`` goes to ``or_else(rest)``,
    or keeps failing with the original error."""

    def handle(cause: Cause[Any]) -> Effect[Any, Any, Any]:
        found = cause.find_recoverable_error()
        if not isinstance(found, Fail):
            return _core._FailCause(found)
        match filter(found.error):
            case Ok(value):
                return f(value)
            case Err(rest):
                return _core._FailCause(cause) if or_else is None else or_else(rest)

    return _core._MatchCause(as_effect(self), handle, None)

catch_cause(f) -> Effect[A_co | B, E2, R_co | R2]

def catch_cause[B, E2 = Never, R2 = Never](
    self, f: Callable[[Cause[E_co]], Effect[B, E2, R2]]
) -> Effect[A_co | B, E2, R_co | R2]:
    """Recover from any failure, including defects. Interruption still
    wins: an interrupted fiber skips this handler."""
    return _core._MatchCause(as_effect(self), f, None)

catch_cause_if(predicate, f) -> Effect[A_co | B, E_co | E2, R_co | R2]

def catch_cause_if[B, E2 = Never, R2 = Never](
    self,
    predicate: Callable[[Cause[E_co]], bool],
    f: Callable[[Cause[E_co]], Effect[B, E2, R2]],
) -> Effect[A_co | B, E_co | E2, R_co | R2]:
    """``catch_cause`` for the causes that pass ``predicate``."""
    return _core._MatchCause(
        as_effect(self), lambda c: f(c) if predicate(c) else _core._FailCause(c), None
    )

catch_cause_filter(filter, f) -> Effect[A_co | B, X | E2, R_co | R2]

def catch_cause_filter[EB, X, B, E2 = Never, R2 = Never](
    self,
    filter: Callable[[Cause[E_co]], Result[EB, Cause[X]]],
    f: Callable[[EB, Cause[E_co]], Effect[B, E2, R2]],
) -> Effect[A_co | B, X | E2, R_co | R2]:
    """Recover with ``f(value, cause)`` from a failure ``filter`` maps to
    ``Ok(value)``; one it maps to ``Err(rest)`` fails with ``rest``."""

    def handle(cause: Cause[E_co]) -> Effect[Any, Any, Any]:
        match filter(cause):
            case Ok(value):
                return f(value, cause)
            case Err(rest):
                return _core._FailCause(rest)

    return _core._MatchCause(as_effect(self), handle, None)

catch_defect(f) -> Effect[A_co | B, E_co | E2, R_co | R2]

def catch_defect[B, E2 = Never, R2 = Never](
    self, f: Callable[[BaseException], Effect[B, E2, R2]]
) -> Effect[A_co | B, E_co | E2, R_co | R2]:
    """Recover from the first defect. Typed errors pass through."""

    def on_failure(cause: Cause[E_co]) -> Effect[Any, Any, Any]:
        defect = cause.find_defect()
        return _core._FailCause(cause) if defect is None else f(defect)

    return _core._MatchCause(as_effect(self), on_failure, None)

or_else_succeed(f) -> Effect[A_co | B, Never, R_co]

def or_else_succeed[B](self, f: Callable[[E_co], B]) -> Effect[A_co | B, Never, R_co]:
    """Succeed with ``f(error)`` instead of failing."""
    return self.catch_all(lambda e: _core._Sync(lambda: f(e)))

eventually() -> Effect[A_co, Never, R_co]

def eventually(self) -> Effect[A_co, Never, R_co]:
    """Run again after each typed failure, yielding in between, until it
    succeeds."""
    loop: Effect[A_co, Never, R_co] = self.catch_all(
        lambda _: _core._FlatMap(_core._YieldNow(), lambda _: loop)
    )
    return loop

flip() -> Effect[E_co, A_co, R_co]

def flip(self) -> Effect[E_co, A_co, R_co]:
    """Swap the error and the value: fail with the value, succeed with
    the error. Defects and interruptions stay failures."""
    return _core._MatchCause(as_effect(self), _recovering(_core._Succeed), _fail)

sandbox() -> Effect[A_co, Cause[E_co], R_co]

def sandbox(self) -> Effect[A_co, Cause[E_co], R_co]:
    """Fail with the whole ``Cause``, defects and interruptions included,
    as the typed error."""
    return _core._MatchCause(as_effect(self), _fail, None)

or_die() -> Effect[A_co, Never, R_co]

def or_die(self) -> Effect[A_co, Never, R_co]:
    """Die with the first typed error instead of failing with it."""
    return self.catch_all(lambda e: _core._FailCause(die_cause(_as_exception(e))))

tap_error(f) -> Effect[A_co, E_co | E2, R_co | R2]

def tap_error[E2 = Never, R2 = Never](
    self, f: Callable[[E_co], Effect[object, E2, R2]]
) -> Effect[A_co, E_co | E2, R_co | R2]:
    def on_failure(cause: Cause[E_co]) -> Effect[Any, Any, Any]:
        found = cause.find_fail()
        if not isinstance(found, Fail):
            return _core._FailCause(cause)
        return _core._FlatMap(f(found.error), lambda _: _core._FailCause(cause))

    return _core._MatchCause(as_effect(self), on_failure, None)

tap_cause(f) -> Effect[A_co, E_co | E2, R_co | R2]

def tap_cause[E2 = Never, R2 = Never](
    self, f: Callable[[Cause[E_co]], Effect[object, E2, R2]]
) -> Effect[A_co, E_co | E2, R_co | R2]:
    return _core._MatchCause(
        as_effect(self), lambda c: _core._FlatMap(f(c), lambda _: _core._FailCause(c)), None
    )

tap_cause_if(predicate, f) -> Effect[A_co, E_co | E2, R_co | R2]

def tap_cause_if[E2 = Never, R2 = Never](
    self,
    predicate: Callable[[Cause[E_co]], bool],
    f: Callable[[Cause[E_co]], Effect[object, E2, R2]],
) -> Effect[A_co, E_co | E2, R_co | R2]:
    """``tap_cause`` for the causes that pass ``predicate``."""

    def on_failure(cause: Cause[E_co]) -> Effect[Any, Any, Any]:
        if not predicate(cause):
            return _core._FailCause(cause)
        return _core._FlatMap(f(cause), lambda _: _core._FailCause(cause))

    return _core._MatchCause(as_effect(self), on_failure, None)

tap_cause_filter(filter, f) -> Effect[A_co, E_co | E2, R_co | R2]

def tap_cause_filter[EB, E2 = Never, R2 = Never](
    self,
    filter: Callable[[Cause[E_co]], Result[EB, object]],
    f: Callable[[EB, Cause[E_co]], Effect[object, E2, R2]],
) -> Effect[A_co, E_co | E2, R_co | R2]:
    """Run ``f(value, cause)`` for a failure ``filter`` maps to
    ``Ok(value)``, then keep failing with the cause."""

    def on_failure(cause: Cause[E_co]) -> Effect[Any, Any, Any]:
        match filter(cause):
            case Ok(value):
                return _core._FlatMap(f(value, cause), lambda _: _core._FailCause(cause))
            case Err():
                return _core._FailCause(cause)

    return _core._MatchCause(as_effect(self), on_failure, None)

tap_defect(f) -> Effect[A_co, E_co | E2, R_co | R2]

def tap_defect[E2 = Never, R2 = Never](
    self, f: Callable[[BaseException], Effect[object, E2, R2]]
) -> Effect[A_co, E_co | E2, R_co | R2]:
    """Run ``f(defect)`` for the first defect, then keep failing."""

    def on_failure(cause: Cause[E_co]) -> Effect[Any, Any, Any]:
        defect = cause.find_defect()
        if defect is None:
            return _core._FailCause(cause)
        return _core._FlatMap(f(defect), lambda _: _core._FailCause(cause))

    return _core._MatchCause(as_effect(self), on_failure, None)

tap_error_tag(tag, f) -> Effect[Any, Any, Any]

def tap_error_tag(
    self,
    tag: type[Any] | tuple[type[Any], ...],
    f: Callable[[Any], Effect[object, Any, Any]],
) -> Effect[Any, Any, Any]:
    """``tap_error`` for the errors that are instances of ``tag`` (a
    class or a tuple of classes)."""
    return self.tap_error(lambda e: f(e) if isinstance(e, tag) else _core._Succeed(None))

ignore(*, log=False, message=None) -> Effect[None, Never, R_co]

def ignore(
    self, *, log: bool | Severity = False, message: str | None = None
) -> Effect[None, Never, R_co]:
    """Discard the value and any typed error. With ``log``, log the
    failure's cause first, after ``message`` if given, at the current
    log level (``True``) or the level given."""
    if log is False:
        return _core._MatchCause(as_effect(self), _ignore_failure, _succeed_none)
    write = _cause_logger(log, message)

    def on_failure(cause: Cause[Any]) -> Effect[None]:
        found = cause.find_recoverable_error()
        return write(cause) if isinstance(found, Fail) else _core._FailCause(found)

    return _core._MatchCause(as_effect(self), on_failure, _succeed_none)

ignore_cause(*, log=False, message=None) -> Effect[None, Never, R_co]

def ignore_cause(
    self, *, log: bool | Severity = False, message: str | None = None
) -> Effect[None, Never, R_co]:
    """Discard the value and the whole cause: typed errors, defects and
    interrupts. ``log`` and ``message`` as for ``ignore``."""
    on_failure = _succeed_none if log is False else _cause_logger(log, message)
    return _core._MatchCause(as_effect(self), on_failure, _succeed_none)

with_error_reporting(*, defects_only=False) -> Effect[A_co, E_co, R_co]

def with_error_reporting(self, *, defects_only: bool = False) -> Effect[A_co, E_co, R_co]:
    """Send the cause of a failure to the ``CurrentErrorReporters``
    (with ``defects_only``, only a cause with a defect), then fail with
    it as before."""
    from pyfect.error_reporter import report_unsafe

    def report(fiber: Fiber[Any, Any], cause: Cause[Any]) -> Effect[None]:
        report_unsafe(fiber, cause, defects_only=defects_only)
        return _core._Succeed(None)

    return as_effect(self).on_error(
        lambda cause: _core._WithFiber(lambda fiber: report(fiber, cause))
    )

pipe(f) -> B

def pipe[B](self, f: Callable[[Self], B], /) -> B:
    """Apply ``f`` to this effect. One step per call, on purpose:
    pyright only narrows types correctly for a single step, so chain
    ``.pipe(a).pipe(b)`` rather than passing several functions."""
    return f(self)

and_then(that) -> Effect[B, E_co | E2, R_co | R2]

def and_then[B, E2 = Never, R2 = Never](
    self, that: Effect[B, E2, R2]
) -> Effect[B, E_co | E2, R_co | R2]:
    """Run ``that`` after this effect, keeping only its value."""
    require_effect(that, "that")
    return _FlatMap(self, lambda _: that)

tap(f) -> Effect[A_co, E_co | E2, R_co | R2]

def tap[E2 = Never, R2 = Never](
    self, f: Callable[[A_co], Effect[object, E2, R2]]
) -> Effect[A_co, E_co | E2, R_co | R2]:
    """Run ``f(value)`` for its effects, keeping the original value."""
    return _FlatMap(self, lambda a: _Map(f(a), lambda _: a))

as_(value) -> Effect[B, E_co, R_co]

def as_[B](self, value: B) -> Effect[B, E_co, R_co]:
    return _Map(self, lambda _: value)

as_void() -> Effect[None, E_co, R_co]

def as_void(self) -> Effect[None, E_co, R_co]:
    return _Map(self, _to_none)

zip(that, *, concurrent=False) -> Effect[tuple[A_co, B], E_co | E2, R_co | R2]

def zip[B, E2 = Never, R2 = Never](
    self, that: Effect[B, E2, R2], *, concurrent: bool = False
) -> Effect[tuple[A_co, B], E_co | E2, R_co | R2]:
    """Pair the results: in order, or both at once with ``concurrent``."""
    require_effect(that, "that")
    return self.zip_with(that, _pair, concurrent=concurrent)

zip_with(that, f, *, concurrent=False) -> Effect[C, E_co | E2, R_co | R2]

def zip_with[B, C, E2 = Never, R2 = Never](
    self, that: Effect[B, E2, R2], f: Callable[[A_co, B], C], *, concurrent: bool = False
) -> Effect[C, E_co | E2, R_co | R2]:
    """Combine the results with ``f``. With ``concurrent``, both run at
    once and the first failure interrupts the other."""
    require_effect(that, "that")
    if not concurrent:
        return _FlatMap(self, lambda a: _Map(that, lambda b: f(a, b)))
    from pyfect._all import all

    return all(self, that, concurrency=2).map(lambda ab: f(*ab))

replicate_effect(n, *, concurrency=None, discard=False) -> Effect[Any, E_co, R_co]

def replicate_effect(
    self, n: int, *, concurrency: Concurrency = None, discard: bool = False
) -> Effect[Any, E_co, R_co]:
    """Run this effect ``n`` times, like ``for_each``, collecting the
    values (or none, with ``discard``)."""
    from pyfect._concurrency import for_each

    repeated = for_each([self] * n, _identity, concurrency=concurrency)
    return repeated.as_void() if discard else repeated

forever() -> Effect[Never, E_co, R_co]

def forever(self) -> Effect[Never, E_co, R_co]:
    loop: Effect[Never, E_co, R_co] = _Suspend(lambda: _FlatMap(self, lambda _: loop))
    return loop

flatten() -> Effect[B, E_co | E2, R_co | R2]

def flatten[B, E2, R2](
    self: Effect[Effect[B, E2, R2], E_co, R_co],
) -> Effect[B, E_co | E2, R_co | R2]:
    """Run the effect this effect succeeds with."""
    return _FlatMap(self, _identity)

head() -> Effect[B, E_co | NoSuchElementException, R_co]

def head[B](
    self: Effect[Iterable[B], E_co, R_co],
) -> Effect[B, E_co | NoSuchElementException, R_co]:
    """The first item of the iterable this effect succeeds with; a
    ``NoSuchElementException`` if it is empty."""

    def first(items: Iterable[B]) -> Effect[B, NoSuchElementException]:
        from pyfect.errors import NoSuchElementException

        for item in items:
            return _Succeed(item)
        return _fail(NoSuchElementException())

    return _FlatMap(self, first)

when(condition) -> Effect[Option[A_co], E_co | E2, R_co | R2]

def when[E2 = Never, R2 = Never](
    self, condition: Effect[bool, E2, R2]
) -> Effect[Option[A_co], E_co | E2, R_co | R2]:
    """Run this effect only if ``condition`` succeeds with ``True``:
    ``Some(value)`` if it ran, ``None`` if it did not."""
    require_effect(condition, "condition")
    from pyfect.option import Some

    return _FlatMap(condition, lambda run: _Map(self, Some) if run else _Succeed(None))

filter_or_fail(predicate, or_fail_with=None) -> Effect[Any, Any, Any]

def filter_or_fail(
    self, predicate: Callable[[Any], bool], or_fail_with: Callable[[Any], Any] | None = None
) -> Effect[Any, Any, Any]:
    """Fail with ``or_fail_with(value)`` (by default a
    ``NoSuchElementException``) when the value fails ``predicate``."""
    if or_fail_with is None:
        from pyfect.errors import NoSuchElementException

        return self.filter_or_else(predicate, lambda _: _fail(NoSuchElementException()))
    return self.filter_or_else(predicate, lambda a: _fail(or_fail_with(a)))

filter_or_else(predicate, or_else) -> Effect[Any, Any, Any]

def filter_or_else(
    self, predicate: Callable[[Any], bool], or_else: Callable[[Any], Effect[Any, Any, Any]]
) -> Effect[Any, Any, Any]:
    """Continue with ``or_else(value)`` when the value fails ``predicate``."""
    return _FlatMap(self, lambda a: _Succeed(a) if predicate(a) else or_else(a))