MemoMap

__init__(parent=None) -> None

def __init__(self, parent: MemoMap | None = None) -> None:
    self._parent = parent
    self._entries: dict[Layer[Any, Any, Any], _Entry] = {}

fork() -> MemoMap

def fork(self) -> MemoMap:
    return MemoMap(self)

get(layer, scope) -> Effect[Context[O], E, R] | None

def get[O, E, R](
    self, layer: Layer[O, E, R], scope: CloseableScope
) -> Effect[Context[O], E, R] | None:
    """The shared build of ``layer``, if this map or a parent has one.
    ``scope`` uses it from now on, whether or not the effect runs."""
    require_layer(layer, "layer")
    entry = self._entries.get(layer)
    if entry is not None:
        entry.observe(scope)
        return entry.deferred.wait()
    return self._parent.get(layer, scope) if self._parent is not None else None

get_or_else_memoize(layer, scope, build) -> Effect[Context[O], E, R]

def get_or_else_memoize[O, E, R](
    self,
    layer: Layer[O, E, R],
    scope: CloseableScope,
    build: Callable[[MemoMap, CloseableScope], Effect[Context[O], E, R]],
) -> Effect[Context[O], E, R]:
    """Reuse ``layer``'s shared build, or run ``build`` as it."""
    require_layer(layer, "layer")

    def run() -> Effect[Context[O], E, R]:
        published: list[Deferred[Context, Any]] = []

        def reuse_or_build() -> Effect[Context[O], E, R]:
            existing = self.get(layer, scope)
            if existing is not None:
                return existing
            entry = _Entry(self, layer)
            if not entry.observe(scope):
                return build(self, scope)
            published.append(entry.deferred)
            self._entries[layer] = entry
            return build(self, entry.scope)

        def settle(ex: Exit[Any, Any]) -> Effect[None]:
            for deferred in published:
                deferred.unsafe_done(ex)
            return _VOID

        # ``settle`` is in place before the entry is published, so even
        # an interrupted first requester completes it for the others.
        return _Suspend(reuse_or_build).on_exit(settle)

    return _Suspend(run)