result_cache.py 901 B

12345678910111213141516171819202122232425262728293031
  1. from collections import OrderedDict
  2. from threading import RLock
  3. from typing import Optional
  4. class ResultCache:
  5. """Simple thread-safe LRU cache for storing small binary results (e.g., PNG bytes)."""
  6. def __init__(self, max_items: int = 64):
  7. self._lock = RLock()
  8. self._store: OrderedDict[str, bytes] = OrderedDict()
  9. self._max_items = max_items
  10. def get(self, key: str) -> Optional[bytes]:
  11. with self._lock:
  12. value = self._store.get(key)
  13. if value is not None:
  14. self._store.move_to_end(key)
  15. return value
  16. def set(self, key: str, value: bytes) -> None:
  17. with self._lock:
  18. self._store[key] = value
  19. self._store.move_to_end(key)
  20. if len(self._store) > self._max_items:
  21. self._store.popitem(last=False)
  22. GLOBAL_RESULT_CACHE = ResultCache(max_items=96)