LayerMap

__init__(rc_map) -> None

def __init__(self, rc_map: RcMap[K, Context[I], E]) -> None:
    self.rc_map = rc_map

make(lookup, *, idle_time_to_live=None, preload_keys=None) -> Effect[LayerMap[Any, Any, Any], Any, Any]

@staticmethod
def make(
    lookup: Callable[[Any], Layer[Any, Any, Any]],
    *,
    idle_time_to_live: IdleTimeToLive[Any] | None = None,
    preload_keys: Iterable[Any] | None = None,
) -> Effect[LayerMap[Any, Any, Any], Any, Any]:
    """A map building ``lookup(key)`` per key (with the services in scope
    now, sharing what enclosing builds built). Released when the
    current scope closes."""

    def body():
        context: Context = yield from _WithFiber(lambda fiber: _Succeed(fiber._context))
        memo_map = fork_or_create(context)

        def build(key: Any) -> Effect[Context, Any, Any]:
            return Scope().flat_map(
                lambda scope: lookup(key).build_with_memo_map(memo_map, scope)
            )

        rc_map = yield from RcMap.make(build, idle_time_to_live=idle_time_to_live)
        for key in preload_keys or ():
            if rc_map._idle_time_to_live(key) != 0:
                yield from scoped(rc_map.get(key))
        return LayerMap(rc_map)

    return _Gen(body)

from_record(layers, *, idle_time_to_live=None, preload=False) -> Effect[LayerMap[Any, Any, Any], Any, Any]

@staticmethod
def from_record(
    layers: Mapping[Any, Layer[Any, Any, Any]],
    *,
    idle_time_to_live: IdleTimeToLive[Any] | None = None,
    preload: bool = False,
) -> Effect[LayerMap[Any, Any, Any], Any, Any]:
    """``make`` over a fixed mapping of layers; ``preload`` preloads
    every key."""
    return LayerMap.make(
        layers.__getitem__,
        idle_time_to_live=idle_time_to_live,
        preload_keys=list(layers) if preload else None,
    )

get(key) -> Layer[I, E]

def get(self, key: K) -> Layer[I, E]:
    """A layer giving the services for ``key``, building them if needed."""
    return effect_context(self.rc_map.get(key))

context_effect(key) -> Effect[Context[I], E, Scope]

def context_effect(self, key: K) -> Effect[Context[I], E, Scope]:
    """The services for ``key``, built if needed and held until the
    current scope closes."""
    return self.rc_map.get(key)

context_effect_option(key) -> Effect[Option[Context[I]], E, Scope]

def context_effect_option(self, key: K) -> Effect[Option[Context[I]], E, Scope]:
    """The services for ``key`` if they are cached (or being built), held
    until the current scope closes; ``None`` otherwise, without
    building anything."""
    return self.rc_map.get_option(key)

invalidate(key) -> Effect[None]

def invalidate(self, key: K) -> Effect[None]:
    """Forget ``key``'s services, so the next use rebuilds them."""
    return self.rc_map.invalidate(key)