Spaces:
Sleeping
Sleeping
File size: 1,705 Bytes
bcceb77 | 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 52 53 54 55 56 57 58 59 | """Simple disk-based caching with TTL support."""
import hashlib
import json
import logging
import os
import pickle
import time
from typing import Any, Optional
logger = logging.getLogger(__name__)
class CacheManager:
"""File-based cache with TTL."""
def __init__(self, cache_dir: str = "data/cache", default_ttl: int = 3600):
self.cache_dir = cache_dir
self.default_ttl = default_ttl
os.makedirs(cache_dir, exist_ok=True)
def _key_to_path(self, key: str) -> str:
hashed = hashlib.md5(key.encode()).hexdigest()
return os.path.join(self.cache_dir, f"{hashed}.pkl")
def get(self, key: str, ttl: Optional[int] = None) -> Optional[Any]:
"""Get cached value if it exists and hasn't expired."""
path = self._key_to_path(key)
if not os.path.exists(path):
return None
ttl = ttl or self.default_ttl
age = time.time() - os.path.getmtime(path)
if age > ttl:
os.remove(path)
return None
try:
with open(path, "rb") as f:
return pickle.load(f)
except Exception:
return None
def set(self, key: str, value: Any) -> None:
"""Cache a value to disk."""
path = self._key_to_path(key)
try:
with open(path, "wb") as f:
pickle.dump(value, f)
except Exception as e:
logger.error(f"Cache write failed for {key}: {e}")
def clear(self) -> None:
"""Clear all cached data."""
for f in os.listdir(self.cache_dir):
path = os.path.join(self.cache_dir, f)
if f.endswith(".pkl"):
os.remove(path)
|