Tiered cache: diskcache FanoutCache plus cachetools LRU: working versions (tested 2026-10-16)

diskcache==5.6.3
cachetools==7.2.2
pytest==9.1.1

from __future__ import annotations

import threading
from collections.abc import Iterator, MutableMapping
from typing import Any, Callable

from cachetools import LRUCache
from cachetools.keys import hashkey
from diskcache import FanoutCache

MISSING = object()


class TieredCache(MutableMapping):
    """L1 cachetools.LRUCache promoted from an L2 diskcache.FanoutCache."""

    def __init__(self, directory: str, maxsize: int = 128, shards: int = 4, size_limit: int = 1 << 24):
        self.memory: LRUCache = LRUCache(maxsize=maxsize)
        self.disk: FanoutCache = FanoutCache(directory, shards=shards, size_limit=size_limit)
        self._lock = threading.RLock()
        self.stats = {"memory": 0, "disk": 0, "miss": 0}

    def __getitem__(self, key: Any) -> Any:
        with self._lock:
            if key in self.memory:
                self.stats["memory"] += 1
                return self.memory[key]
        value = self.disk.get(key, default=MISSING)
        if value is MISSING:
            self.stats["miss"] += 1
            raise KeyError(key)
        with self._lock:
            self.memory[key] = value
        self.stats["disk"] += 1
        return value

    def __setitem__(self, key: Any, value: Any) -> None:
        self.set(key, value)

    def __delitem__(self, key: Any) -> None:
        with self._lock:
            self.memory.pop(key, None)
        del self.disk[key]

    def __iter__(self) -> Iterator[Any]:
        return iter(self.disk)

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

    def set(self, key: Any, value: Any, expire: float | None = None) -> None:
        with self._lock:
            self.memory[key] = value
        self.disk.set(key, value, expire=expire)

    def hits(self) -> tuple[int, int]:
        return self.stats["memory"], self.stats["disk"]

    def close(self) -> None:
        self.disk.close()

    def __enter__(self) -> "TieredCache":
        return self

    def __exit__(self, *exc: object) -> None:
        self.close()


def memoize(cache: MutableMapping, key: Callable[..., Any] = hashkey):
    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            cache_key = key(*args, **kwargs)
            try:
                return cache[cache_key]
            except KeyError:
                value = func(*args, **kwargs)
                cache[cache_key] = value
                return value

        wrapper.cache = cache  # type: ignore[attr-defined]
        return wrapper

    return decorator

import time

import pytest
from cachetools import cached

from tiered import TieredCache, memoize


def test_cached_memory_hit_skips_disk(tmp_path):
    with TieredCache(tmp_path, maxsize=8) as cache:
        calls = []

        @cached(cache)
        def square(n):
            calls.append(n)
            return n * n

        assert square(4) == 16
        assert square(4) == 16
        assert calls == [4]
        assert cache.stats["memory"] == 1
        assert cache.stats["disk"] == 0


def test_lru_eviction_falls_back_to_disk_and_repromotes(tmp_path):
    with TieredCache(tmp_path, maxsize=1) as cache:
        cache["a"] = 1
        cache["b"] = 2
        assert "a" not in cache.memory
        assert cache["a"] == 1
        assert "a" in cache.memory
        assert "b" not in cache.memory
        assert cache.stats["disk"] == 1


def test_stored_none_is_a_hit_but_missing_uses_default(tmp_path):
    with TieredCache(tmp_path, maxsize=8) as cache:
        cache["k"] = None
        cache.memory.clear()
        assert cache.get("k", "fallback") is None
        assert cache.get("missing", "fallback") == "fallback"
        assert cache.stats["miss"] == 1


def test_disk_tier_survives_reopen(tmp_path):
    first = TieredCache(tmp_path, maxsize=2)
    first["k"] = {"n": 1}
    first.close()

    second = TieredCache(tmp_path, maxsize=2)
    try:
        assert dict(second.memory) == {}
        assert second["k"] == {"n": 1}
        assert second.stats["disk"] == 1
    finally:
        second.close()


def test_memory_tier_ignores_disk_ttl(tmp_path):
    with TieredCache(tmp_path, maxsize=8) as cache:
        cache.set("k", "v", expire=0.05)
        time.sleep(0.12)
        assert cache.disk.get("k") is None
        assert cache["k"] == "v"


def test_memoize_reuses_disk_after_memory_eviction(tmp_path):
    with TieredCache(tmp_path, maxsize=1) as cache:
        calls = []

        @memoize(cache)
        def add(x, y):
            calls.append((x, y))
            return x + y

        assert add(2, 3) == 5
        cache["unrelated"] = 1
        assert add(2, 3) == 5
        assert calls == [(2, 3)]
        assert cache.stats["disk"] == 1

[pytest]
pythonpath = .