Spaces:
Paused
Paused
| """Simple in-memory TTL cache for thumbnail results.""" | |
| import time | |
| import asyncio | |
| from typing import Any, Optional | |
| from collections import OrderedDict | |
| class TTLCache: | |
| def __init__(self, max_size: int = 500, ttl_seconds: int = 300): | |
| self.max_size = max_size | |
| self.ttl = ttl_seconds | |
| self._store: OrderedDict[str, tuple[Any, float]] = OrderedDict() | |
| self._lock = asyncio.Lock() | |
| async def get(self, key: str) -> Optional[Any]: | |
| async with self._lock: | |
| if key not in self._store: | |
| return None | |
| value, ts = self._store[key] | |
| if time.time() - ts > self.ttl: | |
| del self._store[key] | |
| return None | |
| # move to end (LRU) | |
| self._store.move_to_end(key) | |
| return value | |
| async def set(self, key: str, value: Any) -> None: | |
| async with self._lock: | |
| if key in self._store: | |
| self._store.move_to_end(key) | |
| self._store[key] = (value, time.time()) | |
| if len(self._store) > self.max_size: | |
| self._store.popitem(last=False) | |
| async def invalidate(self, key: str) -> None: | |
| async with self._lock: | |
| self._store.pop(key, None) | |
| def size(self) -> int: | |
| return len(self._store) | |
| # Global cache instance | |
| thumbnail_cache = TTLCache(max_size=500, ttl_seconds=300) | |