Source code for pyphi.cache.policy

"""Cache policy Protocol and adapters.

A ``CachePolicy`` is the uniform observability and control surface across all of PyPhi's
cache flavors: it declares only ``name``, ``info()``, and ``clear()``. ``get`` / ``put``
/ ``key`` are not part of it; their signatures differ across cache flavors.
"""

from __future__ import annotations

from collections.abc import Callable
from dataclasses import dataclass
from typing import Any
from typing import Protocol
from typing import runtime_checkable

from .cache_utils import _CacheInfo




@runtime_checkable
class CachePolicy(Protocol):
    """Uniform observability + control surface for caches.

    A policy may additionally set a true ``persistent`` attribute to mark
    itself as a durable store; registry-wide ``clear_all()`` skips those.
    """

    name: str

    def info(self) -> _CacheInfo: ...
    def clear(self) -> None: ...



@dataclass
class _DictCacheAdapter:
    """Adapter wrapping a backing dict with externally-tracked hit/miss counts.

    Used by the module-level ``@cache(...)`` decorator and by ``ContentCache`` instances.
    The ``stats`` callable returns ``(hits, misses)``; the wrapper closure that updates
    the counts owns them. The optional ``weigh`` callable returns ``(nbytes, evictions)``
    for caches that track occupancy in bytes; caches that do not report zero for both.
    The optional ``reset`` callable clears the owning cache through its own clear path —
    resetting its tracked byte weight and any latched admission budget along with the
    entries, since the purpose of clearing is recovering memory. Without it, ``clear()``
    falls back to emptying the backing dict, which suits plain unweighted dicts only.
    """

    name: str
    backing: dict[Any, Any]
    stats: Callable[[], tuple[int, int]]
    weigh: Callable[[], tuple[int, int]] | None = None
    reset: Callable[[], None] | None = None

    def info(self) -> _CacheInfo:
        hits, misses = self.stats()
        nbytes, evictions = self.weigh() if self.weigh else (0, 0)
        return _CacheInfo(hits, misses, len(self.backing), nbytes, evictions)

    def clear(self) -> None:
        if self.reset is not None:
            self.reset()
        else:
            self.backing.clear()