layer

program.pipe(provide(DatabaseLive))         # Effect, minus Database
UsersLive.pipe(provide(DatabaseLive))       # Layer, its needs satisfied

succeed(tag, service) -> Layer[T]

def succeed[S, T](tag: TagLike[S, T], service: S) -> Layer[T]:
    """Provide ``service``. Not shared: there is nothing to build."""
    return succeed_context(Context.make(tag, service))

succeed_context(ctx) -> Layer[A]

def succeed_context[A](ctx: Context[A]) -> Layer[A]:
    """Provide every service of ``ctx``."""
    built: Effect[Context] = _Succeed(ctx)
    return Layer(lambda _memo, _scope: built)

sync(tag, make) -> Layer[T]

def sync[S, T](tag: TagLike[S, T], make: Callable[[], S]) -> Layer[T]:
    """Provide ``make()``, called once per build."""
    return sync_context(lambda: Context.make(tag, make()))

sync_context(make) -> Layer[A]

def sync_context[A](make: Callable[[], Context[A]]) -> Layer[A]:
    """Provide every service of ``make()``, called once per build."""
    built: Effect[Context] = sync_effect(make)
    return from_build_memo(lambda _memo, _scope: built)

effect(tag, make) -> Layer[Any, Any, Any]

def effect(tag: TagLike[Any, Any], make: Effect[Any, Any, Any]) -> Layer[Any, Any, Any]:
    """Build a service with an effect. ``make`` may acquire resources
    (it runs with a ``Scope``); they are released with the layer."""
    require_effect(make, "make")
    return effect_context(make.map(lambda service: Context.make(tag, service)))

effect_discard(make) -> Layer[Any, Any, Any]

def effect_discard(make: Effect[object, Any, Any]) -> Layer[Any, Any, Any]:
    """Run ``make`` for its effects (and resources) when the layer builds.
    Builds no services."""
    require_effect(make, "make")
    return effect_context(make.as_(Context.empty()))

suspend(make) -> Layer[O, E, R]

def suspend[O, E = Never, R = Never](make: Callable[[], Layer[O, E, R]]) -> Layer[O, E, R]:
    """The layer ``make()`` returns, made when building. Shared like any
    other layer, so ``make`` runs once per build."""
    return from_build_memo(lambda memo, scope: _Suspend(lambda: make()._build(memo, scope)))

unwrap(eff) -> Layer[Any, Any, Any]

def unwrap(eff: Effect[Layer[Any, Any, Any], Any, Any]) -> Layer[Any, Any, Any]:
    """Build the layer that ``eff`` makes. ``eff`` runs once per build, in
    the layer's scope."""
    require_effect(eff, "eff")
    holder: Layer[_Unwrapped, Any, Any] = effect(_Unwrapped, eff)
    return holder.flat_map(lambda ctx: ctx.get(_Unwrapped))

fresh(layer) -> Layer[O, E, R]

def fresh[O, E, R](layer: Layer[O, E, R]) -> Layer[O, E, R]:
    """A copy of ``layer`` that is not shared: each use builds it, and
    everything under it, anew."""
    require_layer(layer, "layer")
    return Layer(lambda _memo, scope: layer._build(MemoMap(), scope))

mock(tag, **members) -> Layer[T]

def mock[S, T](tag: TagLike[S, T], /, **members: object) -> Layer[T]:
    """Provide a service made of ``members``. Any other attribute is an
    effect that dies with ``UnimplementedError``, and a function returning
    that effect, so a test mocks only what it uses."""
    return succeed(tag, _Mock(tag_class(tag).key, members))  # pyright: ignore[reportArgumentType]

merge(*layers) -> Layer[Any, Any, Any]

def merge(*layers: Layer[Any, Any, Any]) -> Layer[Any, Any, Any]:
    """Build all layers, concurrently, and provide everything they build.
    Their resources are released concurrently too."""
    require_layers(layers, "layers")

    def build(memo: MemoMap, scope: CloseableScope) -> Effect[Context, Any, Any]:
        parallel = scope._fork_now("parallel")
        built = for_each(
            layers,
            lambda layer: layer._build(memo, parallel._fork_now()),
            concurrency="unbounded",
        )
        return built.map(lambda contexts: Context.merge_all(*contexts))

    return from_build_memo(build)

provide(source, /, *, local=False) -> _Provide[Any, Any, Any] | _ProvideLocal[Any, Any, Any] | _ProvideContext[Any]

def provide(
    source: Layer[Any, Any, Any] | Context[Any], /, *, local: bool = False
) -> _Provide[Any, Any, Any] | _ProvideLocal[Any, Any, Any] | _ProvideContext[Any]:
    """Satisfy what a layer builds, for an effect or another layer, or
    provide a context's services to an effect. Use with ``.pipe``. For a
    layer target, the result builds only the target's services;
    ``provide_merge`` keeps both.

    For an effect, layers an enclosing ``provide`` already built are
    reused; ``local=True`` builds everything anew (effects only)."""
    if isinstance(source, Context):
        return _ProvideContext(source)
    require_layer(source, "source", "a Layer or a Context")
    return _ProvideLocal(source) if local else _Provide(source)

provide_merge(layer) -> _ProvideMerge[RO, E2, RI]

def provide_merge[RO, E2 = Never, RI = Never](
    layer: Layer[RO, E2, RI],
) -> _ProvideMerge[RO, E2, RI]:
    """Like ``provide`` for layers, but the result builds both layers'
    services."""
    require_layer(layer, "layer")
    return _ProvideMerge(layer)

catch_tag(tag, f) -> Any

def catch_tag(
    tag: type[Any] | tuple[type[Any], ...], f: Callable[[Any], Layer[Any, Any, Any]]
) -> Any:
    """Build ``f(error)`` instead when the build fails with an instance of
    ``tag`` (a class or a tuple of classes), and remove those errors::

    LayerLive.pipe(layer.catch_tag(ConnectError, lambda e: Fallback))

    The result builds what the fallback builds, which the layer must build
    too (pyright can't spell Effect's ``ROut & ROut2``; the ``catch_all``
    method keeps the layer's services instead)."""
    return _CatchTag[Any, Any, Any, Any](
        lambda memo, scope: _catching_if(
            lambda e: isinstance(e, tag), lambda e: f(e)._build(memo, scope), None
        )
    )

effect_context(eff) -> Layer[Any, Any, Any]

def effect_context(eff: Effect[Context, Any, Any]) -> Layer[Any, Any, Any]:
    """Build every service of the context ``eff`` makes. ``eff`` may acquire
    resources (it runs with a ``Scope``); they are released with the layer."""
    require_effect(eff, "eff")
    return from_build_memo(lambda _memo, scope: _in_scope(eff, scope))

from_build(build) -> Layer[O, E, R]

def from_build[O, E = Never, R = Never](
    build: Callable[[MemoMap, CloseableScope], Effect[Context[O], E, R]],
) -> Layer[O, E, R]:
    """A layer from its build function. Each build gets a child of the
    scope it is given, closed at once if the build fails."""

    def child_scoped(memo: MemoMap, scope: CloseableScope) -> Effect[Context, Any, Any]:
        def run() -> Effect[Context, Any, Any]:
            layer_scope = scope._fork_now()
            return build(memo, layer_scope).on_exit(
                lambda ex: layer_scope.close(ex) if isinstance(ex, Failure) else _VOID
            )

        return _Suspend(run)

    return Layer(child_scoped)

from_build_memo(build) -> Layer[O, E, R]

def from_build_memo[O, E = Never, R = Never](
    build: Callable[[MemoMap, CloseableScope], Effect[Context[O], E, R]],
) -> Layer[O, E, R]:
    """``from_build``, shared through the memo map: within one build, and
    across builds with the same memo map while a scope still uses it."""
    layer: Layer[O, E, R] = from_build(
        lambda memo, scope: memo.get_or_else_memoize(layer, scope, build)
    )
    return layer

parent_span(span) -> Layer[ParentSpan]

def parent_span(span: AnySpan) -> Layer[ParentSpan]:
    """Provide ``span`` as the parent of the spans made with it."""
    built: Effect[Context] = _Succeed(Context.make(ParentSpan, span))
    return Layer(lambda _memo, _scope: built)

span(name, *, on_end=None, **options) -> Layer[ParentSpan]

def span(
    name: str, *, on_end: OnEnd | None = None, **options: Unpack[SpanOptions]
) -> Layer[ParentSpan]:
    """Provide a new span as the parent of the spans made with it. The span
    ends, and ``on_end`` runs, when the layer's scope closes."""
    from pyfect.layer import effect

    return effect(ParentSpan, _scoped_span(name, on_end, options))