ScopedCache

__init__(lookup, context, capacity, time_to_live) -> None

def __init__(
    self,
    lookup: Callable[[K], Effect[A, E, Any]],
    context: Context,
    capacity: int | None,
    time_to_live: Callable[[Exit[A, E], K], Seconds],
) -> None:
    self._lookup = lookup
    self._context = context
    self._capacity = capacity
    self._time_to_live = time_to_live
    # Least recently used first; None once the cache's scope closed.
    self._entries: dict[K, _Entry[A, E]] | None = {}

make(lookup, *, capacity, time_to_live=None, require_services_at='construction') -> Effect[ScopedCache[Any, Any, Any, Any], Never, Any]

@staticmethod
def make(
    lookup: Callable[[Any], Effect[Any, Any, Any]],
    *,
    capacity: int | None,
    time_to_live: TimeToLive[Any, Any, Any] | None = None,
    require_services_at: RequireServicesAt = "construction",
) -> Effect[ScopedCache[Any, Any, Any, Any], Never, Any]:
    """A cache that looks keys up with ``lookup``, each in a scope of
    its own, closed when the entry leaves the cache, and all of them
    when the current scope closes. Options as ``Cache.make``; the
    lookup's ``Scope`` is the entry's, never a requirement."""
    bound = checked_capacity(capacity)
    ttl = time_to_live_of(time_to_live)

    def body(fiber: Fiber[Any, Any]) -> Effect[ScopedCache[Any, Any, Any, Any]]:
        context = fiber._context
        cache = ScopedCache(lookup, context, bound, ttl)
        closing = context.unsafe_get(Scope).add_finalizer(lambda _: _WithFiber(cache._close))
        return closing.as_(cache)

    return _WithFiber(body)

get(key) -> Effect[A, E, R]

def get(self, key: K) -> Effect[A, E, R]:
    """The value for ``key``: cached, joined while another caller looks
    it up, or looked up now. The resource stays open while the entry
    is cached. When the last waiter leaves a lookup in progress, the
    lookup is interrupted and forgotten."""

    def body(restore: Restore) -> Effect[A, E]:
        return _WithFiber(lambda fiber: self._get(fiber, key, restore))

    return uninterruptible_mask(body)

get_option(key) -> Effect[Option[A], E]

def get_option(self, key: K) -> Effect[Option[A], E]:
    """``Some`` value for a cached key (waiting for a lookup in
    progress), ``None`` for a missing or expired one. Never looks up."""

    def body(restore: Restore) -> Effect[Option[A], E]:
        def run(fiber: Fiber[Any, Any]) -> Effect[Option[A], E]:
            def then(entry: _Entry[A, E] | None) -> Effect[Option[A], E]:
                if entry is None:
                    return _Succeed(None)
                return self._await(key, entry, restore).map(Some)

            return self._find(fiber, key, touch=True).flat_map(then)

        return _WithFiber(run)

    return uninterruptible_mask(body)

get_success(key) -> Effect[Option[A]]

def get_success(self, key: K) -> Effect[Option[A]]:
    """``Some`` value for a key whose lookup has succeeded, else
    ``None``. Never waits."""

    def success(entry: _Entry[A, E] | None) -> Option[A]:
        ex = None if entry is None else entry.exit
        return Some(ex.value) if isinstance(ex, Success) else None

    return _WithFiber(
        lambda fiber: self._find(fiber, key, touch=True).map(success)
    ).uninterruptible()

set(key, value) -> Effect[None]

def set(self, key: K, value: A) -> Effect[None]:
    """Cache ``value`` for ``key`` with the time to live of a successful
    lookup, closing the scope of the entry it replaces."""

    def body(fiber: Fiber[Any, Any]) -> Effect[None]:
        entries = self._entries
        if entries is None:
            return interrupt()
        replaced = entries.get(key)
        ex: Exit[A, E] = Success(value)
        entry = _Entry[A, E]()
        entry.complete(ex)
        entry.expires_at = deadline(clock_of(fiber), self._time_to_live(ex, key))
        entries[key] = entry
        evicting = self._evict_overflow(fiber, entries)
        if replaced is None:
            return evicting
        return replaced.scope.close(_DONE).and_then(evicting)

    return _WithFiber(body).uninterruptible()

has(key) -> Effect[bool]

def has(self, key: K) -> Effect[bool]:
    """Whether ``key`` is cached and unexpired (a lookup in progress
    counts). Does not count as a use; closes an expired entry."""
    return _WithFiber(
        lambda fiber: self._find(fiber, key, touch=False).map(lambda e: e is not None)
    ).uninterruptible()

invalidate(key) -> Effect[None]

def invalidate(self, key: K) -> Effect[None]:
    """Forget ``key`` and close its scope."""

    def body() -> Effect[None]:
        entries = self._entries
        if entries is None:
            return interrupt()
        entry = entries.pop(key, None)
        return _VOID if entry is None else entry.scope.close(_DONE)

    return _Suspend(body).uninterruptible()

invalidate_when(key, predicate) -> Effect[bool]

def invalidate_when(self, key: K, predicate: Callable[[A], bool]) -> Effect[bool]:
    """Forget ``key`` and close its scope if its value (waited for if
    still being looked up) passes ``predicate``. ``False`` if it did
    not, if the lookup failed, or if the entry was replaced meanwhile."""

    def body(restore: Restore) -> Effect[bool]:
        def run(fiber: Fiber[Any, Any]) -> Effect[bool]:
            def then(entry: _Entry[A, E] | None) -> Effect[bool]:
                if entry is None:
                    return _Succeed(False)

                def check(value: A) -> Effect[bool]:
                    entries = self._entries
                    if entries is None or not predicate(value):
                        return _Succeed(False)
                    if entries.get(key) is not entry:
                        return _Succeed(False)
                    del entries[key]
                    return entry.scope.close(_DONE).as_(True)

                waited = self._await(key, entry, restore).flat_map(check)
                return waited.catch_all(lambda _: _Succeed(False))

            return self._find(fiber, key, touch=False).flat_map(then)

        return _WithFiber(run)

    return uninterruptible_mask(body)

refresh(key) -> Effect[A, E, R]

def refresh(self, key: K) -> Effect[A, E, R]:
    """Look ``key`` up again in a new scope. The cached entry keeps
    serving ``get`` until the new lookup ends; then the new outcome
    replaces it and its scope closes. A missing key is looked up as
    ``get`` would. An interrupted refresh changes nothing."""

    def body(restore: Restore) -> Effect[A, E, R]:
        return _WithFiber(lambda fiber: self._refresh(fiber, key, restore))

    return uninterruptible_mask(body)

invalidate_all() -> Effect[None]

def invalidate_all(self) -> Effect[None]:
    """Forget every key and close every entry's scope at once; fails
    with every release failure."""

    def body(fiber: Fiber[Any, Any]) -> Effect[None]:
        entries = self._entries
        return interrupt() if entries is None else _release_all(fiber, entries)

    return _WithFiber(body)

size() -> Effect[int]

def size(self) -> Effect[int]:
    """How many entries the cache holds, expired ones included until
    something reads them; 0 once closed."""
    return _Sync(lambda: 0 if self._entries is None else len(self._entries))

keys() -> Effect[list[K]]

def keys(self) -> Effect[list[K]]:
    """The unexpired keys, least recently used first. Closes the
    expired entries."""
    return self._read(lambda key, _: key)

values() -> Effect[list[A]]

def values(self) -> Effect[list[A]]:
    """The values of the unexpired, successful lookups."""
    return self.entries().map(lambda pairs: [value for _, value in pairs])

entries() -> Effect[list[tuple[K, A]]]

def entries(self) -> Effect[list[tuple[K, A]]]:
    """The keys and values of the unexpired, successful lookups.
    Closes the expired entries."""

    def pair(key: K, entry: _Entry[A, E]) -> tuple[K, A] | None:
        ex = entry.exit
        return (key, ex.value) if isinstance(ex, Success) else None

    return self._read(pair)