Spaces:
Running on Zero
Running on Zero
File size: 1,427 Bytes
36333c5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 | """Thread-safe metadata cache used by orchestration components."""
from __future__ import annotations
from collections import OrderedDict
from threading import RLock
from typing import Generic, TypeVar
K = TypeVar("K")
V = TypeVar("V")
class LRUCache(Generic[K, V]):
"""A minimal bounded least-recently-used cache."""
def __init__(self, capacity: int) -> None:
if capacity < 1:
raise ValueError("capacity must be positive")
self.capacity = capacity
self._items: OrderedDict[K, V] = OrderedDict()
self._lock = RLock()
def get(self, key: K) -> V | None:
with self._lock:
value = self._items.get(key)
if value is not None:
self._items.move_to_end(key)
return value
def put(self, key: K, value: V) -> tuple[K, V] | None:
with self._lock:
self._items[key] = value
self._items.move_to_end(key)
if len(self._items) > self.capacity:
return self._items.popitem(last=False)
return None
def clear(self) -> list[V]:
with self._lock:
values = list(self._items.values())
self._items.clear()
return values
def values(self) -> list[V]:
"""Return a snapshot of cached values without exposing mutable state."""
with self._lock:
return list(self._items.values())
|