diff --git a/src/cache_manager.py b/src/cache_manager.py new file mode 100644 index 0000000..df32aec --- /dev/null +++ b/src/cache_manager.py @@ -0,0 +1,160 @@ +"""Caching system for Code Pattern Search CLI.""" + +import hashlib +from pathlib import Path +from typing import Any, Optional +import diskcache as dc + + +class CacheManager: + """Manager for caching API responses and search results.""" + + def __init__( + self, + cache_dir: Optional[Path] = None, + ttl: int = 3600, + ) -> None: + """Initialize the cache manager.""" + if cache_dir is None: + cache_dir = Path.home() / ".cache" / "code-pattern-search" + + self.cache_dir = Path(cache_dir) + self.cache_dir.mkdir(parents=True, exist_ok=True) + self.ttl = ttl + self._cache: Optional[dc.Cache] = None + + @property + def cache(self) -> dc.Cache: + """Get or create the cache instance.""" + if self._cache is None: + self._cache = dc.Cache( + str(self.cache_dir), + ttl=self.ttl, + disk_min_file_size=1024, + ) + return self._cache + + def _generate_key(self, key: str) -> str: + """Generate a cache key from a string.""" + return hashlib.sha256(key.encode()).hexdigest() + + def get(self, key: str) -> Optional[Any]: + """Get a value from the cache.""" + try: + cache_key = self._generate_key(key) + return self.cache.get(cache_key) + except Exception: + return None + + def set(self, key: str, value: Any) -> bool: + """Set a value in the cache.""" + try: + cache_key = self._generate_key(key) + self.cache.set(cache_key, value, expire=self.ttl) + return True + except Exception: + return False + + def delete(self, key: str) -> bool: + """Delete a value from the cache.""" + try: + cache_key = self._generate_key(key) + del self.cache[cache_key] + return True + except KeyError: + return False + except Exception: + return False + + def clear(self) -> bool: + """Clear all cached data.""" + try: + self.cache.clear() + return True + except Exception: + return False + + def get_stats(self) -> dict[str, Any]: + """Get cache statistics.""" + try: + stats = self.cache.stats() + return { + "size": stats["size"], + "hits": stats["hits"], + "misses": stats["misses"], + "cache_size_mb": self._get_cache_size() / (1024 * 1024), + } + except Exception: + return { + "size": 0, + "hits": 0, + "misses": 0, + "cache_size_mb": 0.0, + } + + def _get_cache_size(self) -> int: + """Get total size of cache in bytes.""" + total = 0 + if self.cache_dir.exists(): + for path in self.cache_dir.rglob("*"): + if path.is_file(): + total += path.stat().st_size + return total + + def get_all(self) -> dict[str, Any]: + """Get all cached entries.""" + try: + entries = {} + for key in self.cache: + try: + value = self.cache[key] + entries[key] = value + except Exception: + continue + return entries + except Exception: + return {} + + def get_by_prefix(self, prefix: str) -> dict[str, Any]: + """Get cached entries with a specific prefix.""" + cache_key_prefix = self._generate_key(prefix) + entries = {} + + for key in self.cache: + if key.startswith(cache_key_prefix): + try: + value = self.cache[key] + entries[key] = value + except Exception: + continue + + return entries + + def cleanup(self) -> int: + """Remove expired cache entries.""" + try: + removed = 0 + for key in list(self.cache): + try: + if self.cache.get(key, default=dc.NOTSET) is dc.NOTSET: + del self.cache[key] + removed += 1 + except Exception: + continue + return removed + except Exception: + return 0 + + def close(self) -> None: + """Close the cache connection.""" + if self._cache is not None: + self._cache.close() + self._cache = None + + def __enter__(self) -> "CacheManager": + """Context manager entry.""" + return self + + def __exit__(self, exc_type, exc_val, exc_tb) -> None: + """Context manager exit.""" + self.close()