Layer

__init__(build) -> None

def __init__(self, build: Builder) -> None:
    self._build = build

pipe(f) -> B

def pipe[B](self, f: Callable[[Self], B], /) -> B:
    return f(self)

build_with_memo_map(memo_map, scope) -> Effect[Context[ROut_contra], E_co, RIn_co]

def build_with_memo_map(
    self, memo_map: MemoMap, scope: CloseableScope
) -> Effect[Context[ROut_contra], E_co, RIn_co]:
    """Build with ``memo_map``, releasing into ``scope``. The memo map is
    ``CurrentMemoMap`` during the build and in the built context."""
    building: Effect[Context, Any, Any] = _Locally(
        _Suspend(lambda: self._build(memo_map, scope)),
        lambda ctx: ctx.add(CurrentMemoMap, memo_map),
    )
    return building.map(lambda built: built.add(CurrentMemoMap, memo_map))

build_with_scope(scope) -> Effect[Context[ROut_contra], E_co, RIn_co]

def build_with_scope(self, scope: CloseableScope) -> Effect[Context[ROut_contra], E_co, RIn_co]:
    """Build into ``scope``, reusing what an enclosing build built."""
    return _WithFiber(
        lambda fiber: self.build_with_memo_map(fork_or_create(fiber._context), scope)
    )

build() -> Effect[Context[ROut_contra], E_co, RIn_co | Scope]

def build(self) -> Effect[Context[ROut_contra], E_co, RIn_co | Scope]:
    """Build into the current ``Scope``, reusing what an enclosing build
    built."""
    return _WithFiber(
        lambda fiber: self.build_with_memo_map(
            fork_or_create(fiber._context), fiber._context.unsafe_get(Scope)
        )
    )

launch() -> Effect[Never, E_co, RIn_co]

def launch(self) -> Effect[Never, E_co, RIn_co]:
    """Build the layer and keep it until interrupted, as for a server."""
    return scoped(self.build().and_then(never))

flat_map(f) -> Layer[O2, E_co | E2, RIn_co | R2]

def flat_map[O2, E2 = Never, R2 = Never](
    self, f: Callable[[Context[ROut_contra]], Layer[O2, E2, R2]]
) -> Layer[O2, E_co | E2, RIn_co | R2]:
    """Build the layer ``f`` makes from this one's services."""
    return from_build(
        lambda memo, scope: self._build(memo, scope).flat_map(
            lambda ctx: f(ctx)._build(memo, scope)
        )
    )

tap(f) -> Layer[Any, Any, Any]

def tap(self, f: Callable[[Context], Effect[object, Any, Any]]) -> Layer[Any, Any, Any]:
    """Run ``f`` on the built services, in the layer's scope."""
    return from_build(
        lambda memo, scope: self._build(memo, scope).flat_map(
            lambda ctx: _in_scope(f(ctx), scope).as_(ctx)
        )
    )

tap_error(f) -> Layer[Any, Any, Any]

def tap_error(self, f: Callable[[Any], Effect[object, Any, Any]]) -> Layer[Any, Any, Any]:
    """Run ``f`` on a typed build error, in the layer's scope; the build
    still fails with it."""
    return from_build(
        lambda memo, scope: self._build(memo, scope).catch_all(
            lambda e: _in_scope(f(e), scope).and_then(_FailCause(fail_cause(e)))
        )
    )

tap_cause(f) -> Layer[Any, Any, Any]

def tap_cause(
    self, f: Callable[[Cause[Any]], Effect[object, Any, Any]]
) -> Layer[Any, Any, Any]:
    """Run ``f`` on the cause of a failed build, in the layer's scope;
    the build still fails with it."""
    return from_build(
        lambda memo, scope: self._build(memo, scope).catch_cause(
            lambda c: _in_scope(f(c), scope).and_then(_FailCause(c))
        )
    )

or_die() -> Layer[ROut_contra, Never, RIn_co]

def or_die(self) -> Layer[ROut_contra, Never, RIn_co]:
    """Die with a typed build error instead of failing with it."""
    return Layer(lambda memo, scope: self._build(memo, scope).or_die())

catch_all(f) -> Layer[O, E2, R | R2]

def catch_all[O, E, R, E2 = Never, R2 = Never](
    self: Layer[O, E, R], f: Callable[[E], Layer[O, E2, R2]]
) -> Layer[O, E2, R | R2]:
    """Build ``f(error)`` instead when the build fails with a typed
    error. The fallback must build at least what this layer builds."""
    return Layer(
        lambda memo, scope: self._build(memo, scope).catch_all(
            lambda e: f(e)._build(memo, scope)
        )
    )

catch_cause(f) -> Layer[O, E2, R | R2]

def catch_cause[O, E, R, E2 = Never, R2 = Never](
    self: Layer[O, E, R], f: Callable[[Cause[E]], Layer[O, E2, R2]]
) -> Layer[O, E2, R | R2]:
    """Build ``f(cause)`` instead when the build fails, defects
    included. The fallback must build at least what this layer builds."""
    return Layer(
        lambda memo, scope: self._build(memo, scope).catch_cause(
            lambda c: f(c)._build(memo, scope)
        )
    )

update_service(tag, f) -> Layer[ROut_contra, E_co, RIn_co | T]

def update_service[S, T](
    self, tag: TagLike[S, T], f: Callable[[S], S]
) -> Layer[ROut_contra, E_co, RIn_co | T]:
    """Build with ``f`` applied to the service of ``tag`` that this
    layer needs (which it then needs from outside)."""
    updated = effect_context(tag().map(lambda service: Context.make(tag, f(service))))
    return feed(updated, self, keep_dep=False)

with_span(name, *, on_end=None, **options) -> Layer[ROut_contra, E_co, RIn_co]

def with_span(
    self,
    name: str,
    *,
    on_end: OnEnd | None = None,
    **options: Unpack[TracedSpanOptions],
) -> Layer[ROut_contra, E_co, RIn_co]:
    """Build inside a new span, the parent of the spans made while
    building, and inside its stack frame (this call site unless
    ``capture_stack_trace`` says otherwise). The span ends, and
    ``on_end`` runs, when the layer's scope closes."""
    from pyfect._layer_tracing import with_span
    from pyfect._stack_frame import stack_of

    stack = stack_of(options.get("capture_stack_trace", True), 1)
    return with_span(self, name, on_end, options, stack)

with_parent_span(span, *, capture_stack_trace=True) -> Layer[ROut_contra, E_co, RIn_co]

def with_parent_span(
    self, span: AnySpan, *, capture_stack_trace: CaptureStackTrace = True
) -> Layer[ROut_contra, E_co, RIn_co]:
    """Build with ``span`` as the parent of the spans made while
    building, and, unless it is external, inside its stack frame."""
    from pyfect._layer_tracing import with_parent_span
    from pyfect._stack_frame import stack_of

    return with_parent_span(self, span, stack_of(capture_stack_trace, 1))