"""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()