pyfect

import pyfect as fx

all(*effects, /, *, concurrency=None, discard=False, mode='default') -> Effect[Any, Any, Any]

def all(
    *effects: Effect[Any, Any, Any],
    concurrency: Concurrency = None,
    discard: bool = False,
    mode: Mode = "default",
) -> Effect[Any, Any, Any]:
    """Run effects and collect their results as a tuple, in order."""
    require_effects(effects, "effects")
    collected = for_each(effects, _step_for(mode), concurrency=concurrency)
    return collected.as_void() if discard else collected.map(tuple)

all_dict(effects, *, concurrency=None, discard=False, mode='default') -> Effect[Any, Any, Any]

def all_dict(
    effects: Mapping[Any, Effect[Any, Any, Any]],
    *,
    concurrency: Concurrency = None,
    discard: bool = False,
    mode: Mode = "default",
) -> Effect[Any, Any, Any]:
    """Run the effects in a mapping and collect their results under the
    same keys, in the mapping's order. The mapping is read each time the
    effect runs."""
    step = _step_for(mode)

    def start() -> Effect[Any, Any, Any]:
        keys = list(effects)
        collected = for_each([effects[k] for k in keys], step, concurrency=concurrency)
        if discard:
            return collected.as_void()
        return collected.map(lambda values: dict(zip(keys, values, strict=True)))

    return _Suspend(start)

for_each(items, f, *, concurrency=None) -> Effect[list[B], E, R]

def for_each[T, B, E = Never, R = Never](
    items: Iterable[T],
    f: Callable[[T], Effect[B, E, R]],
    *,
    concurrency: Concurrency = None,
) -> Effect[list[B], E, R]:
    """Run ``f`` on every item and collect the results in order.

    With ``concurrency``, up to that many run at once. The first failure
    interrupts the rest. ``items`` is read each time the effect runs.
    """
    if concurrency is None:
        return _for_each_sequential(items, f)
    limit_for = _parse_limit(concurrency)

    def start() -> Effect[list[B], E, R]:
        snapshot = list(items)
        limit = limit_for(len(snapshot))
        if limit == 1:
            return _for_each_sequential(snapshot, f)
        return _for_each_concurrent(snapshot, f, limit)

    return _Suspend(start)

race_all(*effects) -> Effect[A, EE, RR]

def race_all[A, EE = Never, RR = Never](*effects: Effect[A, EE, RR]) -> Effect[A, EE, RR]:
    """The first to succeed wins; the rest are interrupted. Fails only if
    all fail."""
    require_effects(effects, "effects")
    if not effects:
        raise ValueError("race_all needs at least one effect")
    return _race(list(effects), first_exit_wins=False)

race_all_first(*effects) -> Effect[A, EE, RR]

def race_all_first[A, EE = Never, RR = Never](*effects: Effect[A, EE, RR]) -> Effect[A, EE, RR]:
    """The first to finish wins, success or failure; the rest are
    interrupted."""
    require_effects(effects, "effects")
    if not effects:
        raise ValueError("race_all_first needs at least one effect")
    return _race(list(effects), first_exit_wins=True)

acquire_async_context(acquire) -> Effect[T, E, R | Scope]

def acquire_async_context[T, E = Never, R = Never](
    acquire: Effect[AbstractAsyncContextManager[T], E, R],
) -> Effect[T, E, R | Scope]:
    """``acquire_context`` for ``async with`` managers. Needs an asyncio
    loop (``run_async`` / ``run_main``), like ``from_async``."""
    require_effect(acquire, "acquire")

    def enter(
        cm: AbstractAsyncContextManager[T],
    ) -> Effect[tuple[AbstractAsyncContextManager[T], T]]:
        return from_async(cm.__aenter__).map(lambda value: (cm, value))

    def release(
        entered: tuple[AbstractAsyncContextManager[T], T], ex: Exit[Any, Any]
    ) -> Effect[None]:
        return from_async(lambda: entered[0].__aexit__(*_exc_info(ex))).map(_discard)

    return acquire_release(acquire.flat_map(enter), release).map(_second)

acquire_context(acquire) -> Effect[T, E, R | Scope]

def acquire_context[T, E = Never, R = Never](
    acquire: Effect[AbstractContextManager[T], E, R],
) -> Effect[T, E, R | Scope]:
    """Enter the context manager ``acquire`` produces; exit it when the
    scope closes. If ``__enter__`` or ``__exit__`` raises, that is a defect."""
    require_effect(acquire, "acquire")

    def enter(cm: AbstractContextManager[T]) -> Effect[tuple[AbstractContextManager[T], T]]:
        return _Sync(lambda: (cm, cm.__enter__()))

    def release(entered: tuple[AbstractContextManager[T], T], ex: Exit[Any, Any]) -> Effect[None]:
        return _Sync(lambda: _discard(entered[0].__exit__(*_exc_info(ex))))

    return acquire_release(acquire.flat_map(enter), release).map(_second)

interruptible_mask(f) -> Effect[A, E, R]

def interruptible_mask[A, E, R](f: Callable[[Restore], Effect[A, E, R]]) -> Effect[A, E, R]:
    """Run ``f`` interruptibly. ``f`` receives ``restore``, which runs an
    inner effect with the interruptibility the caller had. An interrupt that
    arrived while the caller was uninterruptible lands at once."""

    def body(fiber: Fiber[Any, Any]) -> Effect[A, E, R]:
        if fiber._interruptible:
            return f(_keep_interruptibility)
        # As in Effect, ``f`` runs even when an interrupt is already waiting.
        return _SetInterruptible(f(_uninterruptible), True)

    return _WithFiber(body)

uninterruptible_mask(f) -> Effect[A, E, R]

def uninterruptible_mask[A, E, R](f: Callable[[Restore], Effect[A, E, R]]) -> Effect[A, E, R]:
    """Run ``f`` uninterruptibly. ``f`` receives ``restore``, which runs an
    inner effect with the interruptibility the caller had."""

    def body(fiber: Fiber[Any, Any]) -> Effect[A, E, R]:
        restore = _interruptible if fiber._interruptible else _uninterruptible
        return _SetInterruptible(_Suspend(lambda: f(restore)), False)

    return _WithFiber(body)

filter(items, predicate, *, concurrency=None) -> Effect[list[Any], Any, Any]

def filter(
    items: Iterable[Any],
    predicate: Callable[[Any], bool | Effect[bool, Any, Any]],
    *,
    concurrency: Concurrency = None,
) -> Effect[list[Any], Any, Any]:
    """Keep the items that pass ``predicate``, in order. An effectful
    predicate runs like ``for_each``, ``concurrency`` included."""

    def keep(item: Any) -> Effect[tuple[bool, Any], Any, Any]:
        verdict = predicate(item)
        if isinstance(verdict, Effect):
            return verdict.map(lambda passed: (passed, item))
        return _Succeed((verdict, item))

    return for_each(items, keep, concurrency=concurrency).map(
        lambda checked: [item for passed, item in checked if passed]
    )

filter_map(items, filter) -> Effect[list[B]]

def filter_map[A, B](
    items: Iterable[A], filter: Callable[[A], Result[B, object]]
) -> Effect[list[B]]:
    """The values of the items ``filter`` maps to ``Ok(value)``, in order;
    ``Err`` items are dropped. ``items`` is read each time the effect runs."""

    def run() -> Effect[list[B]]:
        return _Succeed([r.value for r in map(filter, items) if isinstance(r, Ok)])

    return _Suspend(run)

filter_map_effect(items, filter, *, concurrency=None) -> Effect[list[B], E, R]

def filter_map_effect[A, B, E = Never, R = Never](
    items: Iterable[A],
    filter: Callable[[A], Effect[Result[B, object], E, R]],
    *,
    concurrency: Concurrency = None,
) -> Effect[list[B], E, R]:
    """``filter_map`` with an effectful filter, run like ``for_each``. The
    values are collected as each filter finishes, so with ``concurrency``
    they come in completion order, as in Effect."""

    def run() -> Effect[list[B], E, R]:
        out: list[B] = []

        def keep(item: A) -> Effect[None, E, R]:
            def record(result: Result[B, object]) -> None:
                if isinstance(result, Ok):
                    out.append(result.value)

            return filter(item).map(record)

        return for_each(items, keep, concurrency=concurrency).as_(out)

    return _Suspend(run)

find_first(items, predicate) -> Effect[Option[A], E, R]

def find_first[A, E = Never, R = Never](
    items: Iterable[A], predicate: Callable[[A], Effect[bool, E, R]]
) -> Effect[Option[A], E, R]:
    """The first item for which ``predicate`` succeeds with ``True``
    (``Some(item)``), checking one at a time and stopping there; ``None``
    if there is none. ``items`` is read each time the effect runs."""
    return find_first_filter(
        items, lambda item: predicate(item).map(lambda keep: Ok(item) if keep else Err(item))
    )

find_first_filter(items, filter) -> Effect[Option[B], E, R]

def find_first_filter[A, B, E = Never, R = Never](
    items: Iterable[A], filter: Callable[[A], Effect[Result[B, object], E, R]]
) -> Effect[Option[B], E, R]:
    """The value of the first item ``filter`` passes (``Ok(value)``), as
    ``Some(value)``, checking one at a time and stopping there; ``None`` if
    every item is rejected (``Err``)."""

    def check(it: Iterator[A]) -> Effect[Option[B], E, R]:
        for item in it:
            return filter(item).flat_map(lambda passed: _found(passed, it))
        return _Succeed(None)

    def _found(passed: Result[B, object], it: Iterator[A]) -> Effect[Option[B], E, R]:
        return _Succeed(Some(passed.value)) if isinstance(passed, Ok) else check(it)

    return _Suspend(lambda: check(iter(items)))

first_success_of(effects) -> Effect[A, EE, RR]

def first_success_of[A, EE = Never, RR = Never](
    effects: Iterable[Effect[A, EE, RR]],
) -> Effect[A, EE, RR]:
    """Run the effects in order until one succeeds. If all fail, fail with
    the last one's error; later effects never run. ``effects`` is read each
    time the effect runs, and an empty one is a defect."""

    def start() -> Effect[A, EE, RR]:
        it = iter(effects)
        first = next(it, None)
        if first is None:
            return _FailCause(die(ValueError("Received an empty collection of effects")))
        return _try_next(first, it)

    return _Suspend(start)

partition(items, f, *, concurrency=None) -> Effect[tuple[list[B], list[E]], Never, R]

def partition[T, B, E = Never, R = Never](
    items: Iterable[T],
    f: Callable[[T], Effect[B, E, R]],
    *,
    concurrency: Concurrency = None,
) -> Effect[tuple[list[B], list[E]], Never, R]:
    """Run ``f`` on every item, like ``for_each``, without stopping at
    failures: the values that succeeded and the typed errors, each in
    order. Defects and interruptions still propagate."""
    return for_each(items, lambda item: f(item).result(), concurrency=concurrency).map(_split)

reduce(items, zero, f) -> Effect[Z, E, R]

def reduce[A, Z, E = Never, R = Never](
    items: Iterable[A], zero: Callable[[], Z], f: Callable[[Z, A], Effect[Z, E, R]]
) -> Effect[Z, E, R]:
    """Fold ``items`` from the left with an effectful ``f``, one at a time,
    stopping at the first failure. ``items`` is read now, as Effect reads
    it; ``zero()`` is called each time the effect runs."""
    elements = list(items)

    def start() -> Effect[Z, E, R]:
        state: Effect[Z, E, R] = _Succeed(zero())
        for item in elements:
            state = _fold(state, f, item)
        return state

    return _Suspend(start)

validate(items, f, *, concurrency=None, discard=False) -> Effect[Any, Any, Any]

def validate(
    items: Iterable[Any],
    f: Callable[[Any], Effect[Any, Any, Any]],
    *,
    concurrency: Concurrency = None,
    discard: bool = False,
) -> Effect[Any, Any, Any]:
    """Like ``partition``, but fail with every typed error (never an empty
    tuple) if any item failed; otherwise succeed with the values, or with
    ``None`` when ``discard`` is set."""

    def decide(split: tuple[list[Any], list[Any]]) -> Effect[Any, Any, Never]:
        passes, fails = split
        if fails:
            return _FailCause(fail((fails[0], *fails[1:])))
        return _Succeed(None if discard else passes)

    return partition(items, f, concurrency=concurrency).flat_map(decide)

catch_if(refinement, f, *, or_else=None) -> Any

def catch_if(
    refinement: Callable[[Any], bool],
    f: Callable[[Any], Effect[Any, Any, Any]],
    *,
    or_else: Callable[[Any], Effect[Any, Any, Any]] | None = None,
) -> Any:
    """Recover from the typed errors ``refinement`` accepts and remove them
    from ``E``. For a plain ``bool`` predicate, use the ``catch_if`` method."""
    handle = _catching_if(refinement, f, or_else)
    if or_else is None:
        return _CatchSome[Any, Any, Any, Any](handle)
    return _CatchAll[Any, Any, Any, Any, Any, Any](handle)

catch_reason(tag, reason, f, *, or_else=None) -> Any

def catch_reason(
    tag: type[Any],
    reason: type[Any],
    f: ReasonHandler[Effect[Any, Any, Any]],
    *,
    or_else: ReasonHandler[Effect[Any, Any, Any]] | None = None,
) -> Any:
    """Recover from a ``tag`` error whose ``reason`` is an instance of
    ``reason``: ``f(reason, error)``. With ``or_else``, the ``tag`` errors
    with any other reason go to ``or_else(reason, error)``. Defects win,
    as in every typed recovery."""
    run = _reason_catch(tag, ((reason, f),), or_else)
    return (
        _CatchReason[Any, Any, Any](run)
        if or_else is None
        else _CatchReasonAll[Any, Any, Any, Any, Any, Any, Any, Any](run)
    )

catch_reasons(tag, *cases, *, or_else=None) -> Any

def catch_reasons(
    tag: type[Any],
    *cases: tuple[type[Any], ReasonHandler[Effect[Any, Any, Any]]],
    or_else: ReasonHandler[Effect[Any, Any, Any]] | None = None,
) -> Any:
    """``catch_reason`` with one handler per reason class: each case is a
    ``(reason_class, handler)`` pair, tried in order (Effect's
    ``catchReasons`` object of handlers)."""
    run = _reason_catch(tag, cases, or_else)
    return (
        _CatchReason[Any, Any, Any](run)
        if or_else is None
        else _CatchReasonAll[Any, Any, Any, Any, Any, Any, Any, Any](run)
    )

catch_tag(tag, f, *, or_else=None) -> Any

def catch_tag(
    tag: type[Any] | tuple[type[Any], ...],
    f: Callable[[Any], Effect[Any, Any, Any]],
    *,
    or_else: Callable[[Any], Effect[Any, Any, Any]] | None = None,
) -> Any:
    """Recover from the errors that are instances of ``tag`` (a class or a
    tuple of classes) and remove them from ``E``::

    eff.pipe(catch_tag(NotFound, lambda e: succeed(None)))
    """
    handle = _catching_if(lambda e: isinstance(e, tag), f, or_else)
    if or_else is None:
        return _CatchSome[Any, Any, Any, Any](handle)
    return _CatchAll[Any, Any, Any, Any, Any, Any](handle)

provide_context(ctx) -> _ProvideContext[RO]

def provide_context[RO](ctx: Context[RO]) -> _ProvideContext[RO]:
    """Provide every service of ``ctx`` and remove them from ``R``::

    eff.pipe(provide_context(ctx))
    """
    return _ProvideContext(ctx)

provide_service_effect(tag, acquire) -> _ProvideServiceEffect[T, E2, R2]

def provide_service_effect[S, T, E2 = Never, R2 = Never](
    tag: TagLike[S, T], acquire: Effect[S, E2, R2]
) -> _ProvideServiceEffect[T, E2, R2]:
    """Run ``acquire`` and provide its value as ``tag``'s service, removing
    ``tag`` from ``R``::

    eff.pipe(provide_service_effect(Database, connect()))
    """
    require_effect(acquire, "acquire")
    return _ProvideServiceEffect(tag_class(tag), acquire)

service_option(tag) -> Effect[Option[S]]

def service_option[S, T](tag: TagLike[S, T]) -> Effect[Option[S]]:
    """``Some(service)`` if it is provided, else ``None``. Never needs the
    service, so ``tag`` does not join ``R``."""
    return _WithFiber(lambda fiber: _Succeed(fiber._context.get_option(tag)))

unwrap_reason(tag) -> _UnwrapReason[T, T]

def unwrap_reason[T: HasReason[Any]](tag: type[T]) -> _UnwrapReason[T, T]:
    """Fail with the ``reason`` of a ``tag`` error instead of the error
    itself, promoting the reasons into ``E``."""
    run = _reason_catch(tag, (), lambda reason, _: _fail(reason))
    return _UnwrapReason(run)

update_service_scoped(tag, f, *, reset=None) -> Effect[None, Never, T | Scope]

def update_service_scoped[S, T](
    tag: TagLike[S, T],
    f: Callable[[S], S],
    *,
    reset: Callable[[S, S, S], S] | None = None,
) -> Effect[None, Never, T | Scope]:
    """Replace the service of ``tag`` with ``f(service)`` for the rest of
    the current fiber's region, and put the original back when the current
    ``Scope`` closes, unless the service has changed again by then.
    ``reset(original, updated, current)`` decides what to put back instead.
    """

    def update(fiber: Fiber[Any, Any]) -> Effect[None]:
        original = fiber._context.unsafe_get(tag)
        updated = f(original)
        fiber._set_context(fiber._context.add(tag, updated))

        def put_back() -> Effect[None]:
            current = fiber._context.get_option(tag)
            if current is None or (reset is None and current.value is not updated):
                return _VOID
            restored = original if reset is None else reset(original, updated, current.value)
            fiber._set_context(fiber._context.add(tag, restored))
            return _VOID

        def finalizer(_ex: Exit[Any, Any]) -> Effect[None]:
            return _Suspend(put_back)

        scope = fiber._context.unsafe_get(Scope)
        if scope._closed is not None:
            return finalizer(scope._closed)
        # Not ``add_finalizer``: it restores the closing fiber's services
        # afterwards, which would undo the put-back when this fiber closes.
        scope._add(object(), finalizer)
        return _VOID

    return _WithFiber(update).uninterruptible()