effect

succeed(value) -> Effect[A]

def succeed[A](value: A) -> Effect[A]:
    return _Succeed(value)

fail(error) -> Effect[Never, E]

def fail[E](error: E) -> Effect[Never, E]:
    return _FailCause(fail_cause_of(error))

fail_cause(cause) -> Effect[Never, E]

def fail_cause[E](cause: Cause[E]) -> Effect[Never, E]:
    return _FailCause(cause)

die(defect) -> Effect[Never]

def die(defect: BaseException) -> Effect[Never]:
    return _FailCause(die_cause(defect))

from_exit(ex) -> Effect[A, E]

def from_exit[A, E](ex: Exit[A, E]) -> Effect[A, E]:
    return _FromExit(ex)

sync(thunk) -> Effect[A]

def sync[A](thunk: Callable[[], A]) -> Effect[A]:
    """Run a side effect when the effect runs. If ``thunk`` raises, that is
    a defect; use ``try_`` to turn exceptions into typed errors."""
    return _Sync(thunk)

fail_sync(thunk) -> Effect[Never, E]

def fail_sync[E](thunk: Callable[[], E]) -> Effect[Never, E]:
    """Fail with ``thunk()``, computed each time the effect runs."""
    return _Suspend(lambda: fail(thunk()))

fail_cause_sync(thunk) -> Effect[Never, E]

def fail_cause_sync[E](thunk: Callable[[], Cause[E]]) -> Effect[Never, E]:
    """Fail with the cause ``thunk()``, computed each time the effect runs."""
    return _Suspend(lambda: fail_cause(thunk()))

from_result(result) -> Effect[A, E]

def from_result[A = Never, E = Never](result: Result[A, E]) -> Effect[A, E]:
    """Succeed with an ``Ok``'s value or fail with an ``Err``'s error."""
    return _Succeed(result.value) if isinstance(result, Ok) else fail(result.error)

from_option(option, on_none=None) -> Effect[Any, Any]

def from_option(option: Option[Any], on_none: Callable[[], Any] | None = None) -> Effect[Any, Any]:
    """Succeed with a ``Some``'s value. ``None`` fails with ``on_none()``,
    computed each time the effect runs, or a ``NoSuchElementException``."""
    if option is not None:
        return _Succeed(option.value)
    if on_none is None:
        return fail(NoSuchElementException("from_option: None"))
    return fail_sync(on_none)

from_optional(value) -> Effect[A, NoSuchElementException]

def from_optional[A = Never](value: A | None) -> Effect[A, NoSuchElementException]:
    """Succeed with ``value``, or fail with ``NoSuchElementException`` if it
    is ``None``. When ``None`` can be a real value, use ``from_option``."""
    return fail(NoSuchElementException()) if value is None else _Succeed(value)

transpose_option(option) -> Effect[Option[A], E, R]

def transpose_option[A = Never, E = Never, R = Never](
    option: Option[Effect[A, E, R]],
) -> Effect[Option[A], E, R]:
    """Run the effect inside a ``Some``, keeping its value in a ``Some``."""
    if option is None:
        return _Succeed(None)
    require_effect(option.value, "option.value")
    return option.value.map(Some)

try_(thunk, catch=None) -> Effect[Any, Any]

def try_(
    thunk: Callable[[], Any], catch: Callable[[Exception], Any] | None = None
) -> Effect[Any, Any]:
    """Run ``thunk``, mapping a raised exception to a typed error with
    ``catch``, or to an ``UnknownError``. If ``catch`` raises, that is a
    defect."""
    to_error = _unknown_error_in("try_") if catch is None else catch

    def run() -> Effect[Any, Any]:
        try:
            return _Succeed(thunk())
        except Exception as exc:
            return _failure_of(to_error, exc)

    return _Suspend(run)

suspend(thunk) -> Effect[A, E, R]

def suspend[A, E = Never, R = Never](thunk: Callable[[], Effect[A, E, R]]) -> Effect[A, E, R]:
    """Build the effect each time it runs. Useful for recursion."""
    return _Suspend(thunk)

callback(register) -> Effect[A, E, R]

def callback[A, E = Never, R = Never](
    register: Callable[[Callable[[Effect[A, E, R]], None]], Effect[object, Never, R] | None],
) -> Effect[A, E, R]:
    """Wrap a callback-based API. ``register`` receives ``resume``; call it
    once, from any thread, with the effect to continue with. Return an
    effect to run if the fiber is interrupted while waiting, or ``None``.
    """
    return _Callback(register)

from_async(make) -> Effect[A]

def from_async[A](make: Callable[[], Awaitable[A]]) -> Effect[A]:
    """Await an awaitable. Requires ``run_async``/``run_main`` (an asyncio
    loop). Raised exceptions are defects; interruption cancels the task and
    waits for it to finish."""
    return _await(make, None)

try_async(make, catch=None) -> Effect[Any, Any]

def try_async(
    make: Callable[[], Awaitable[Any]], catch: Callable[[Exception], Any] | None = None
) -> Effect[Any, Any]:
    """Like ``from_async``, mapping an exception raised by ``make`` or the
    awaitable to a typed error with ``catch``, or to an ``UnknownError``. If
    ``catch`` raises, that is a defect."""
    return _await(make, _unknown_error_in("try_async") if catch is None else catch)

blocking(thunk) -> Effect[A]

def blocking[A](thunk: Callable[[], A]) -> Effect[A]:
    """Run blocking code on a worker thread without blocking other fibers.
    Raised exceptions are defects. A running thread cannot be stopped, so
    interruption abandons it and does not wait."""
    return _blocking(thunk, None)

try_blocking(thunk, catch=None) -> Effect[Any, Any]

def try_blocking(
    thunk: Callable[[], Any], catch: Callable[[Exception], Any] | None = None
) -> Effect[Any, Any]:
    """Like ``blocking``, mapping a raised exception to a typed error with
    ``catch``, or to an ``UnknownError``. If ``catch`` raises, that is a
    defect."""
    return _blocking(thunk, _unknown_error_in("try_blocking") if catch is None else catch)

yield_now() -> Effect[None]

def yield_now() -> Effect[None]:
    """Let other fibers run before continuing."""
    return _YieldNow()

interrupt() -> Effect[Never]

def interrupt() -> Effect[Never]:
    """Interrupt the current fiber."""
    return _WithFiber(lambda f: _FailCause(interrupt_cause(f.id)))

fiber_id() -> Effect[int]

def fiber_id() -> Effect[int]:
    return _WithFiber(lambda f: _Succeed(f.id))

current_fiber() -> Effect[Fiber[Any, Any]]

def current_fiber() -> Effect[Fiber[Any, Any]]:
    return _WithFiber(_Succeed)

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

@gen
def program():
    user = yield from find_user("u1")
    return user.name

def gen[A, E = Never, R = Never](
    f: Callable[[], Generator[Effect[Any, E, R], Any, A]],
) -> Effect[A, E, R]:
    """Build an effect from a generator function. Inside it,
    ``x = yield from eff`` runs ``eff``. Works as a decorator::

        @gen
        def program():
            user = yield from find_user("u1")
            return user.name
    """
    return _Gen(f)

fn(f, **options) -> Callable[..., Effect[Any, Any, Any]] | _Traced

@fn
def find_user(user_id: str):
    db = yield from Database()
    ...

def fn(
    f: Callable[..., Generator[Effect[Any, Any, Any], Any, Any]] | str,
    /,
    **options: Unpack[SpanOptions],
) -> Callable[..., Effect[Any, Any, Any]] | _Traced:
    """Turn a generator function with parameters into a function that
    returns an effect::

        @fn
        def find_user(user_id: str):
            db = yield from Database()
            ...

    ``@fn("name")`` (with span options as keywords) also runs every call
    inside a span of that name, the parent of the spans made inside.

    Each call of a named function also runs inside a stack frame for its
    call site under one for the definition (``"name (definition)"``), so a
    failure inside records both (``tracer.StackTrace``). The bare ``@fn``
    is Effect's ``fnUntraced``: no span and no frames.
    """
    if isinstance(f, str):
        return _Traced(f, options, call_site(1))

    @functools.wraps(f)
    def wrapper(*args: Any, **kwargs: Any) -> Effect[Any, Any, Any]:
        return _Gen(lambda: f(*args, **kwargs))

    return wrapper

catch_no_such_element(eff) -> Effect[Option[Any], Any, Any]

def catch_no_such_element(eff: Effect[Any, Any, Any]) -> Effect[Option[Any], Any, Any]:
    """Succeed with ``None`` on a ``NoSuchElementException`` and with
    ``Some(value)`` on success, removing the error from ``E``::

    eff.pipe(catch_no_such_element)
    """
    require_effect(eff, "eff")
    return eff.match_effect(
        on_failure=lambda e: _Succeed(None) if isinstance(e, NoSuchElementException) else fail(e),
        on_success=lambda a: _Succeed(Some(a)),
    )

provide_service(tag, service) -> _ProvideService[S, T]

def provide_service[S, T](tag: TagLike[S, T], service: S) -> _ProvideService[S, T]:
    """Provide one service directly and remove it from ``R``::

    eff.pipe(provide_service(Database, InMemoryDb()))
    """
    return _ProvideService(tag_class(tag), service)

context(*tags) -> Effect[Context[T], Never, T]

ctx = yield from fx.context(Database, Clock)

def context[T = Never](*tags: TagLike[Any, T]) -> Effect[Context[T], Never, T]:
    """The current fiber's services. Name the tags the caller reads (Effect's
    ``context<R>()``): they join ``R``, and the context is typed to hold
    them::

        ctx = yield from fx.context(Database, Clock)
    """
    return _WithFiber(lambda fiber: _Succeed(fiber._context))