Design an LRU Cache.

# lru_cache.py, part 1: the interfaces
from __future__ import annotations

import threading
from abc import ABC, abstractmethod
from collections.abc import Callable
from typing import Generic, Hashable, TypeVar

K = TypeVar("K", bound=Hashable)
V = TypeVar("V")


class Cache(ABC, Generic[K, V]):
    @abstractmethod
    def get(self, key: K) -> V | None: ...

    @abstractmethod
    def put(self, key: K, value: V) -> None: ...

    @abstractmethod
    def delete(self, key: K) -> bool: ...


class EvictionPolicy(ABC, Generic[K]):
    """Decides who leaves. It only sees keys; the cache owns the values."""

    @abstractmethod
    def on_insert(self, key: K) -> None: ...

    @abstractmethod
    def on_access(self, key: K) -> None: ...

    @abstractmethod
    def on_delete(self, key: K) -> None: ...

    @abstractmethod
    def victim(self) -> K: ...

# part 2: the doubly linked list, with sentinels so there are no None checks at the ends
class Node(Generic[K]):
    __slots__ = ("key", "prev", "next")

    def __init__(self, key: K | None = None) -> None:
        self.key = key
        self.prev: Node[K] = self
        self.next: Node[K] = self


class DoublyLinkedList(Generic[K]):
    def __init__(self) -> None:
        self.head: Node[K] = Node()  # head.next is the most recently used
        self.tail: Node[K] = Node()  # tail.prev is the least recently used
        self.head.next, self.tail.prev = self.tail, self.head

    def push_front(self, node: Node[K]) -> None:
        node.prev, node.next = self.head, self.head.next
        self.head.next.prev = node
        self.head.next = node

    def unlink(self, node: Node[K]) -> None:
        node.prev.next, node.next.prev = node.next, node.prev

    def last(self) -> Node[K] | None:
        return None if self.tail.prev is self.head else self.tail.prev

    def keys(self) -> list[K]:
        out: list[K] = []
        n = self.head.next
        while n is not self.tail:
            out.append(n.key)  # type: ignore[arg-type]  # only sentinels have key None
            n = n.next
        return out

# part 3: LRU as a policy: a dict to find the node in O(1), a list to keep the order
class LRUPolicy(EvictionPolicy[K]):
    def __init__(self) -> None:
        self._nodes: dict[K, Node[K]] = {}
        self._order: DoublyLinkedList[K] = DoublyLinkedList()

    def on_insert(self, key: K) -> None:
        node = Node(key)
        self._nodes[key] = node
        self._order.push_front(node)

    def on_access(self, key: K) -> None:
        node = self._nodes[key]
        self._order.unlink(node)
        self._order.push_front(node)

    def on_delete(self, key: K) -> None:
        self._order.unlink(self._nodes.pop(key))

    def victim(self) -> K:
        node = self._order.last()
        if node is None:
            raise LookupError("nothing to evict")
        return node.key  # type: ignore[return-value]

    def order(self) -> list[K]:
        return self._order.keys()


class FIFOPolicy(EvictionPolicy[K]):
    """A second policy, to prove the seam: reads don't change who leaves first."""

    def __init__(self) -> None:
        self._lru = LRUPolicy[K]()

    def on_insert(self, key: K) -> None:
        self._lru.on_insert(key)

    def on_access(self, key: K) -> None:
        pass

    def on_delete(self, key: K) -> None:
        self._lru.on_delete(key)

    def victim(self) -> K:
        return self._lru.victim()

# part 4: the cache: storage + policy + capacity + one lock
class BoundedCache(Cache[K, V]):
    def __init__(self, capacity: int, policy: EvictionPolicy[K], on_evict: Callable[[K, V], None] | None = None) -> None:
        if capacity < 1:
            raise ValueError("capacity must be at least 1")
        self.capacity = capacity
        self.policy = policy
        self.on_evict = on_evict
        self._data: dict[K, V] = {}
        self._lock = threading.Lock()  # get() reorders the list, so even reads take it
        self.hits = self.misses = 0

    def get(self, key: K) -> V | None:
        with self._lock:
            if key not in self._data:
                self.misses += 1
                return None
            self.hits += 1
            self.policy.on_access(key)
            return self._data[key]

    def put(self, key: K, value: V) -> None:
        evicted: tuple[K, V] | None = None
        with self._lock:
            if key in self._data:  # update: no eviction, but it counts as a use
                self._data[key] = value
                self.policy.on_access(key)
                return
            if len(self._data) >= self.capacity:
                old = self.policy.victim()
                self.policy.on_delete(old)
                evicted = (old, self._data.pop(old))
            self._data[key] = value
            self.policy.on_insert(key)
        if evicted and self.on_evict:
            self.on_evict(*evicted)  # outside the lock: a slow listener must not stall the cache

    def delete(self, key: K) -> bool:
        with self._lock:
            if key not in self._data:
                return False
            del self._data[key]
            self.policy.on_delete(key)
            return True

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

# part 5: the interviewer's test sequence, then a thread hammer
if __name__ == "__main__":
    lru: LRUPolicy[str] = LRUPolicy()
    cache: BoundedCache[str, str] = BoundedCache(3, lru, on_evict=lambda k, v: print("evicted", k))
    cache.put("A", "Aarav")
    cache.put("B", "Bhavna")
    cache.put("C", "Chetan")
    print(lru.order())  # ['C', 'B', 'A']
    cache.get("A")  # A is now the most recent
    cache.put("D", "Diya")  # evicted B
    print(lru.order())  # ['D', 'A', 'C']
    cache.put("C", "Chetan v2")  # update, no eviction
    print(lru.order(), cache.get("B"))  # ['C', 'D', 'A'] None

    fifo: BoundedCache[str, int] = BoundedCache(2, FIFOPolicy())
    fifo.put("x", 1)
    fifo.put("y", 2)
    fifo.get("x")
    fifo.put("z", 3)  # FIFO ignores the read: x leaves anyway
    print(fifo.get("x"), fifo.get("y"), fifo.get("z"))  # None 2 3

    shared: BoundedCache[int, int] = BoundedCache(100, LRUPolicy())

    def worker(seed: int) -> None:
        for i in range(2000):
            k = (seed * 7919 + i * 31) % 300
            if shared.get(k) is None:
                shared.put(k, k * k)

    threads = [threading.Thread(target=worker, args=(s,)) for s in range(8)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    print(len(shared), "entries,", shared.hits + shared.misses, "lookups")  # 100 entries, 16000 lookups

['C', 'B', 'A']
evicted B
['D', 'A', 'C']
['C', 'D', 'A'] None
None 2 3
100 entries, 16000 lookups