File size: 1,783 Bytes
3ab01a7 | 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 50 51 | 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])
|