64 lines
1.8 KiB
Python
64 lines
1.8 KiB
Python
"""Simple in-memory cache for expensive operations."""
|
|
import time
|
|
import hashlib
|
|
import json
|
|
from typing import Any, Optional, Callable
|
|
import threading
|
|
|
|
class Cache:
|
|
"""Thread-safe in-memory cache with TTL."""
|
|
|
|
def __init__(self):
|
|
self._cache: dict = {}
|
|
self._lock = threading.Lock()
|
|
|
|
def get(self, key: str) -> Optional[Any]:
|
|
"""Get value from cache if not expired."""
|
|
with self._lock:
|
|
if key not in self._cache:
|
|
return None
|
|
|
|
entry = self._cache[key]
|
|
if entry['expires'] < time.time():
|
|
del self._cache[key]
|
|
return None
|
|
|
|
return entry['value']
|
|
|
|
def set(self, key: str, value: Any, ttl: int = 60):
|
|
"""Set value in cache with TTL (seconds)."""
|
|
with self._lock:
|
|
self._cache[key] = {
|
|
'value': value,
|
|
'expires': time.time() + ttl
|
|
}
|
|
|
|
def delete(self, key: str):
|
|
"""Delete key from cache."""
|
|
with self._lock:
|
|
if key in self._cache:
|
|
del self._cache[key]
|
|
|
|
def clear(self):
|
|
"""Clear all cache."""
|
|
with self._lock:
|
|
self._cache.clear()
|
|
|
|
def get_or_set(self, key: str, factory: Callable, ttl: int = 60) -> Any:
|
|
"""Get from cache or call factory and cache result."""
|
|
value = self.get(key)
|
|
if value is not None:
|
|
return value
|
|
|
|
value = factory()
|
|
self.set(key, value, ttl)
|
|
return value
|
|
|
|
# Global cache instance
|
|
cache = Cache()
|
|
|
|
def make_key(*args, **kwargs) -> str:
|
|
"""Create cache key from arguments."""
|
|
key_data = json.dumps({'args': args, 'kwargs': kwargs}, sort_keys=True, default=str)
|
|
return hashlib.md5(key_data.encode()).hexdigest()
|