Cache API Reference¶

class Cache(collections.abc.MutableMapping):
    """Mutable mapping to serve as a simple cache or cache base class.

    Subclasses must override ``popitem`` to implement a specific eviction
    policy.  The base class evicts arbitrary items (dict order).

    Args:
        maxsize: Maximum capacity (item count or weighted size).
        getsizeof: Optional callable returning the size of a value.
            Defaults to 1 per item.
    """

    __marker = object()

    def __init__(
        self,
        maxsize: int | float,
        getsizeof: Callable[[Any], int] | None = None,
    ):
        if maxsize < 0:
            raise ValueError("maxsize must be non-negative")
        self.__data: dict[Any, Any] = {}
        self.__currsize: int | float = 0
        self.__maxsize = maxsize
        if getsizeof is not None:
            self.__size: dict | _DefaultSize = {}
            self.getsizeof = getsizeof  # type: ignore[assignment]
        else:
            self.__size = _DefaultSize()

    def __repr__(self) -> str:
        cls = self.__class__.__name__
        return (
            f"{cls}({self.__data!r}, "
            f"maxsize={self.__maxsize}, currsize={self.__currsize})"
        )

    def __getitem__(self, key: Any) -> Any:
        try:
            return self.__data[key]
        except KeyError:
            return self.__missing__(key)

    def __setitem__(self, key: Any, value: Any) -> None:
        maxsize = self.__maxsize
        size = self.getsizeof(value)
        if size > maxsize:
            raise ValueError("value too large")
        if key in self.__data:
            diffsize = size - self.__size[key]
        else:
            diffsize = size
        while self.__currsize + diffsize > maxsize:
            self.popitem()
        if key in self.__data:
            # Key might have been evicted during popitem loop
            self.__currsize -= self.__size[key]
        self.__data[key] = value
        self.__size[key] = size
        self.__currsize += size

    def __delitem__(self, key: Any) -> None:
        size = self.__size.pop(key)
        del self.__data[key]
        self.__currsize -= size

    def __contains__(self, key: object) -> bool:
        return key in self.__data

    def __missing__(self, key: Any) -> Any:
        raise KeyError(key)

    def __iter__(self):
        return iter(self.__data)

    def __len__(self) -> int:
        return len(self.__data)

    @staticmethod
    def getsizeof(value: Any) -> int:
        """Return the size of *value*.  Defaults to 1."""
        return 1

    @property
    def maxsize(self) -> int | float:
        """Maximum cache capacity."""
        return self.__maxsize

    @property
    def currsize(self) -> int | float:
        """Current cache size (sum of item sizes)."""
        return self.__currsize

    def get(self, key: Any, default: Any = None) -> Any:
        if key in self:
            return self[key]
        return default

    def pop(self, key: Any, default: Any = __marker) -> Any:
        if key in self:
            value = self[key]
            del self[key]
            return value
        if default is self.__marker:
            raise KeyError(key)
        return default

    def setdefault(self, key: Any, default: Any = None) -> Any:
        if key in self:
            return self[key]
        try:
            self[key] = default
        except ValueError:
            pass
        return default

    def popitem(self) -> tuple[Any, Any]:
        """Remove and return an arbitrary ``(key, value)`` pair.

        Subclasses override this to implement eviction policies.
        """
        try:
            key = next(iter(self.__data))
        except StopIteration:
            raise KeyError(f"{type(self).__name__} is empty") from None
        return key, self.pop(key)

    def clear(self) -> None:
        self.__data.clear()
        self.__size.clear()
        self.__currsize = 0

class FIFOCache(Cache):
    """First In First Out (FIFO) cache implementation.

    Evicts the oldest inserted item when the cache is full.  Accessing an
    item does **not** change its eviction priority.
    """

    def __init__(
        self,
        maxsize: int | float,
        getsizeof: Callable[[Any], int] | None = None,
    ):
        super().__init__(maxsize, getsizeof)
        self.__order: collections.OrderedDict[Any, None] = collections.OrderedDict()

    def __setitem__(self, key: Any, value: Any) -> None:
        if key in self:
            del self.__order[key]
        super().__setitem__(key, value)
        self.__order[key] = None

    def __delitem__(self, key: Any) -> None:
        super().__delitem__(key)
        del self.__order[key]

    def popitem(self) -> tuple[Any, Any]:
        """Remove and return the oldest inserted ``(key, value)`` pair."""
        try:
            key = next(iter(self.__order))
        except StopIteration:
            raise KeyError(f"{type(self).__name__} is empty") from None
        return key, self.pop(key)

    def clear(self) -> None:
        super().clear()
        self.__order.clear()

class LRUCache(Cache):
    """Least Recently Used (LRU) cache implementation.

    Evicts the least recently accessed item when the cache is full.
    Both reads and writes count as accesses.
    """

    def __init__(
        self,
        maxsize: int | float,
        getsizeof: Callable[[Any], int] | None = None,
    ):
        super().__init__(maxsize, getsizeof)
        self.__order: collections.OrderedDict[Any, None] = collections.OrderedDict()

    def __getitem__(self, key: Any) -> Any:
        # Bypass super().__getitem__ to avoid MRO dispatch overhead;
        # call Cache.__getitem__ directly + inline move_to_end.
        value = Cache.__getitem__(self, key)
        self.__order.move_to_end(key)
        return value

    def __setitem__(self, key: Any, value: Any) -> None:
        Cache.__setitem__(self, key, value)
        try:
            self.__order.move_to_end(key)
        except KeyError:
            self.__order[key] = None

    def __delitem__(self, key: Any) -> None:
        Cache.__delitem__(self, key)
        del self.__order[key]

    def popitem(self) -> tuple[Any, Any]:
        """Remove and return the least recently used ``(key, value)`` pair."""
        try:
            key = next(iter(self.__order))
        except StopIteration:
            raise KeyError(f"{type(self).__name__} is empty") from None
        return key, self.pop(key)

    def clear(self) -> None:
        Cache.clear(self)
        self.__order.clear()

class LFUCache(Cache):
    """Least Frequently Used (LFU) cache implementation.

    Evicts the least frequently accessed item when the cache is full.
    All operations are O(1).
    """

    def __init__(
        self,
        maxsize: int | float,
        getsizeof: Callable[[Any], int] | None = None,
    ):
        super().__init__(maxsize, getsizeof)
        self.__root = _LFUNode()  # sentinel
        self.__links: dict[Any, _LFUNode] = {}

    def __getitem__(self, key: Any) -> Any:
        value = super().__getitem__(key)
        self.__touch(key)
        return value

    def __setitem__(self, key: Any, value: Any) -> None:
        existed = key in self
        super().__setitem__(key, value)
        if existed:
            self.__touch(key)
        else:
            # New key starts at count=1
            root = self.__root
            first = root.next
            if first is root or first.count != 1:
                node = _LFUNode(count=1)
                root.insert_after(node)
            else:
                node = first
            node.keys.add(key)
            self.__links[key] = node

    def __delitem__(self, key: Any) -> None:
        super().__delitem__(key)
        node = self.__links.pop(key)
        node.keys.discard(key)
        if not node.keys:
            node.unlink()

    def __touch(self, key: Any) -> None:
        """Increment frequency count for *key*."""
        node = self.__links[key]
        new_count = node.count + 1
        nxt = node.next
        if nxt is self.__root or nxt.count != new_count:
            new_node = _LFUNode(count=new_count)
            node.insert_after(new_node)
        else:
            new_node = nxt
        new_node.keys.add(key)
        node.keys.discard(key)
        self.__links[key] = new_node
        if not node.keys:
            node.unlink()

    def popitem(self) -> tuple[Any, Any]:
        """Remove and return the least frequently used ``(key, value)`` pair."""
        root = self.__root
        node = root.next
        if node is root:
            raise KeyError(f"{type(self).__name__} is empty")
        key = next(iter(node.keys))
        # Bypass self.pop → self[key] → __touch to avoid wasteful
        # frequency increment on a key that is about to be deleted.
        value = Cache.__getitem__(self, key)
        del self[key]
        return key, value

    def clear(self) -> None:
        super().clear()
        self.__links.clear()
        self.__root.prev = self.__root
        self.__root.next = self.__root

class TTLCache(LRUCache):
    """LRU cache with per-item time-to-live (TTL).

    Items are evicted when they expire (checked lazily on access) or when
    the cache is full (LRU order).

    Args:
        maxsize: Maximum cache capacity.
        ttl: Time-to-live in seconds for each item.
        timer: Callable returning the current time (default: ``time.monotonic``).
        getsizeof: Optional callable returning the size of a value.
    """

    def __init__(
        self,
        maxsize: int | float,
        ttl: float,
        *,
        timer: Callable[[], float] = time.monotonic,
        getsizeof: Callable[[Any], int] | None = None,
    ):
        LRUCache.__init__(self, maxsize, getsizeof)
        self.__ttl = ttl
        self.__timer = _Timer(timer)
        self.__root = _TTLLink()  # sentinel for expiry list
        self.__links: dict[Any, _TTLLink] = {}

    @property
    def ttl(self) -> float:
        """Time-to-live in seconds."""
        return self.__ttl

    @property
    def timer(self) -> _Timer:
        """The timer function."""
        return self.__timer

    def __contains__(self, key: object) -> bool:
        link = self.__links.get(key)
        if link is None:
            return False
        return self.__timer() < link.expires

    def __getitem__(self, key: Any) -> Any:
        link = self.__links.get(key)
        if link is not None and self.__timer() >= link.expires:
            del self[key]
            return self.__missing__(key)
        # Bypass LRUCache.__getitem__ → Cache.__getitem__ call chain;
        # inline both to save one method-call and one dict lookup.
        # Same direct-call pattern used by LFUCache.popitem.
        value = Cache.__getitem__(self, key)
        self._LRUCache__order.move_to_end(key)  # type: ignore[attr-defined]
        return value

    def __setitem__(self, key: Any, value: Any) -> None:
        with self.__timer as now:
            self.expire(now)
            LRUCache.__setitem__(self, key, value)
            # Update or create expiry link
            link = self.__links.get(key)
            if link is not None:
                link.unlink()
            else:
                link = _TTLLink()
                self.__links[key] = link
            link.key = key
            link.expires = now + self.__ttl
            # Append to end of expiry list (newest)
            tail = self.__root.prev
            link.prev = tail
            link.next = self.__root
            tail.next = link
            self.__root.prev = link

    def __delitem__(self, key: Any) -> None:
        LRUCache.__delitem__(self, key)
        link = self.__links.pop(key)
        link.unlink()

    def __iter__(self):
        now = self.__timer()
        for key in super().__iter__():
            link = self.__links.get(key)
            if link is not None and now < link.expires:
                yield key

    def __len__(self) -> int:
        now = self.__timer()
        # Single dict lookup via .get() instead of ``key in`` + ``[key]``.
        links_get = self.__links.get
        return sum(
            1
            for key in super().__iter__()
            if (link := links_get(key)) is not None and now < link.expires
        )

    @property
    def currsize(self) -> int | float:
        """Current size, excluding expired items."""
        self.expire()
        return super().currsize

    def expire(self, time: float | None = None) -> list[tuple[Any, Any]]:
        """Remove expired items and return them as a list of ``(key, value)`` pairs."""
        if time is None:
            time = self.__timer()
        root = self.__root
        expired: list[tuple[Any, Any]] = []
        link = root.next
        while link is not root and link.expires <= time:
            nxt = link.next
            key = link.key
            try:
                value = Cache.__getitem__(self, key)
                expired.append((key, value))
                del self[key]
            except KeyError:
                pass
            link = nxt
        return expired

    def clear(self) -> None:
        LRUCache.clear(self)
        self.__links.clear()
        self.__root.prev = self.__root
        self.__root.next = self.__root