| from __future__ import annotations |
|
|
| from collections import OrderedDict |
| from threading import Lock |
| import time |
|
|
|
|
| class ImageBytesCache: |
| """Small TTL/LRU accelerator, never a replacement for durable originals.""" |
|
|
| def __init__(self, *, max_bytes: int, max_entries: int, ttl: float = 300.0): |
| self.max_bytes = max_bytes |
| self.max_entries = max_entries |
| self.ttl = ttl |
| self._lock = Lock() |
| self._items: OrderedDict[str, tuple[float, bytes]] = OrderedDict() |
| self._bytes = 0 |
|
|
| def get(self, key: str) -> bytes | None: |
| with self._lock: |
| item = self._items.get(key) |
| if item is None: |
| return None |
| expires, payload = item |
| if time.monotonic() >= expires: |
| self._items.pop(key) |
| self._bytes -= len(payload) |
| return None |
| self._items.move_to_end(key) |
| return payload |
|
|
| def put(self, key: str, payload: bytes) -> None: |
| if not payload or len(payload) > self.max_bytes: |
| self.discard(key) |
| return |
| with self._lock: |
| previous = self._items.pop(key, None) |
| if previous is not None: |
| self._bytes -= len(previous[1]) |
| self._items[key] = (time.monotonic() + self.ttl, payload) |
| self._bytes += len(payload) |
| while self._bytes > self.max_bytes or len(self._items) > self.max_entries: |
| _, (_expires, evicted) = self._items.popitem(last=False) |
| self._bytes -= len(evicted) |
|
|
| def discard(self, key: str) -> None: |
| with self._lock: |
| previous = self._items.pop(key, None) |
| if previous is not None: |
| self._bytes -= len(previous[1]) |
|
|
