Cache

__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.
    self._entries: dict[K, _Entry[A, E]] = {}

make(lookup, *, capacity, time_to_live=None, require_services_at='construction') -> Effect[Cache[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[Cache[Any, Any, Any, Any], Never, Any]:
    """A cache that looks keys up with ``lookup``, keeping at most
    ``capacity`` entries (``None``: no limit). ``time_to_live`` is a
    duration, or a function of the lookup's ``Exit`` and the key; the
    default is forever, and zero keeps nothing. Lookups see the
    services of the context made in, overridden by the caller's;
    ``require_services_at="lookup"`` makes ``get`` require them."""
    bound = checked_capacity(capacity)
    ttl = time_to_live_of(time_to_live)
    return _WithFiber(lambda fiber: _Succeed(Cache(lookup, fiber._context, bound, ttl)))

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. A lookup runs on a fiber of its own, so it
    lives on while anyone waits for it; when the last waiter leaves, it
    is interrupted and forgotten."""
    return _WithFiber(lambda fiber: self._get(fiber, key))

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(fiber: Fiber[Any, Any]) -> Effect[Option[A], E]:
        entry = self._find(fiber, key, touch=True)
        return _Succeed(None) if entry is None else entry.await_from(fiber).map(Some)

    return _WithFiber(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 body(fiber: Fiber[Any, Any]) -> Effect[Option[A]]:
        entry = self._find(fiber, key, touch=True)
        ex = None if entry is None else entry.fiber.unsafe_poll()
        return _Succeed(Some(ex.value) if isinstance(ex, Success) else None)

    return _WithFiber(body)

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."""

    def body(fiber: Fiber[Any, Any]) -> Effect[None]:
        ex: Exit[A, E] = Success(value)
        ttl = self._time_to_live(ex, key)
        if ttl == 0:
            self._discard(key)
            return _VOID
        entry = _Entry[A, E](_finished(fiber, ex))
        entry.expires_at = deadline(clock_of(fiber), ttl)
        self._entries[key] = entry
        self._evict_overflow()
        return _VOID

    return _WithFiber(body)

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."""
    return _WithFiber(lambda fiber: _Succeed(self._find(fiber, key, touch=False) is not None))

invalidate(key) -> Effect[None]

def invalidate(self, key: K) -> Effect[None]:
    """Forget ``key``. A lookup in progress goes on for its waiters."""
    return _Sync(lambda: self._discard(key))

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

def invalidate_when(self, key: K, predicate: Callable[[A], bool]) -> Effect[bool]:
    """Forget ``key`` 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(fiber: Fiber[Any, Any]) -> Effect[bool]:
        entry = self._find(fiber, key, touch=False)
        if entry is None:
            return _Succeed(False)

        def check(value: A) -> bool:
            if not predicate(value) or self._entries.get(key) is not entry:
                return False
            del self._entries[key]
            return True

        return entry.await_from(fiber).map(check).catch_cause(lambda _: _Succeed(False))

    return _WithFiber(body)

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

def refresh(self, key: K) -> Effect[A, E, R]:
    """Look ``key`` up again. A cached value keeps serving ``get`` until
    the new lookup succeeds or fails; a missing key is cached at once,
    as ``get`` would."""
    return _WithFiber(lambda fiber: self._refresh(fiber, key))

invalidate_all() -> Effect[None]

def invalidate_all(self) -> Effect[None]:
    """Forget every key."""
    return _Sync(self._entries.clear)

size() -> Effect[int]

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

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

def keys(self) -> Effect[list[K]]:
    """The unexpired keys, least recently used first. Drops the
    expired ones."""
    return _WithFiber(lambda fiber: _Succeed([key for key, _ in self._live(fiber)]))

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. Drops
    the expired entries."""

    def body(fiber: Fiber[Any, Any]) -> Effect[list[tuple[K, A]]]:
        pairs: list[tuple[K, A]] = []
        for key, entry in self._live(fiber):
            ex = entry.fiber.unsafe_poll()
            if isinstance(ex, Success):
                pairs.append((key, ex.value))
        return _Succeed(pairs)

    return _WithFiber(body)