"""Caching layer for expensive operations."""
import json
import time
from typing import Any, Optional, Dict
from pathlib import Path
import logging
logger = logging.getLogger(__name__)
class CacheManager:
"""Simple file-based cache with TTL support."""
def __init__(self, cache_dir: str = "./cache", default_ttl: int = 3600):
"""Initialize cache manager.
Args:
cache_dir: Directory to store cache files
default_ttl: Default time-to-live in seconds
"""
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(parents=True, exist_ok=True)
self.default_ttl = default_ttl
self._cache: Dict[str, Dict[str, Any]] = {}
def _get_cache_file(self, key: str) -> Path:
"""Get cache file path for a key.
Args:
key: Cache key
Returns:
Path to cache file
"""
# Create a safe filename from key
safe_key = "".join(c if c.isalnum() or c in ('-', '_') else '_' for c in key)
return self.cache_dir / f"{safe_key}.cache"
def get(self, key: str) -> Optional[Any]:
"""Get value from cache.
Args:
key: Cache key
Returns:
Cached value if exists and not expired, None otherwise
"""
# Check in-memory cache first
if key in self._cache:
entry = self._cache[key]
if time.time() < entry['expires_at']:
logger.debug(f"Cache hit (memory): {key}")
return entry['value']
else:
del self._cache[key]
# Check file cache
cache_file = self._get_cache_file(key)
if cache_file.exists():
try:
with open(cache_file, 'r') as f:
entry = json.load(f)
if time.time() < entry['expires_at']:
# Load into memory cache
self._cache[key] = entry
logger.debug(f"Cache hit (file): {key}")
return entry['value']
else:
# Expired, remove file
cache_file.unlink()
except Exception as e:
logger.error(f"Error reading cache file: {e}")
logger.debug(f"Cache miss: {key}")
return None
def set(self, key: str, value: Any, ttl: Optional[int] = None) -> None:
"""Set value in cache.
Args:
key: Cache key
value: Value to cache
ttl: Time-to-live in seconds (uses default if not specified)
"""
ttl = ttl or self.default_ttl
expires_at = time.time() + ttl
entry = {
'value': value,
'expires_at': expires_at,
'created_at': time.time(),
}
# Save to memory
self._cache[key] = entry
# Save to file
cache_file = self._get_cache_file(key)
try:
with open(cache_file, 'w') as f:
json.dump(entry, f)
logger.debug(f"Cached: {key} (TTL: {ttl}s)")
except Exception as e:
logger.error(f"Error writing cache file: {e}")
def delete(self, key: str) -> None:
"""Delete value from cache.
Args:
key: Cache key to delete
"""
# Remove from memory
if key in self._cache:
del self._cache[key]
# Remove file
cache_file = self._get_cache_file(key)
if cache_file.exists():
cache_file.unlink()
logger.debug(f"Cache deleted: {key}")
def clear(self) -> None:
"""Clear all cache entries."""
# Clear memory
self._cache.clear()
# Clear files
for cache_file in self.cache_dir.glob("*.cache"):
cache_file.unlink()
logger.info("Cache cleared")
def clean_expired(self) -> None:
"""Remove expired cache entries."""
current_time = time.time()
# Clean memory cache
expired_keys = [
key for key, entry in self._cache.items()
if current_time >= entry['expires_at']
]
for key in expired_keys:
del self._cache[key]
# Clean file cache
for cache_file in self.cache_dir.glob("*.cache"):
try:
with open(cache_file, 'r') as f:
entry = json.load(f)
if current_time >= entry['expires_at']:
cache_file.unlink()
except Exception:
# If we can't read it, delete it
cache_file.unlink()
logger.debug(f"Cleaned {len(expired_keys)} expired cache entries")
