Initial upload with CI/CD workflow

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