""" Caching functionality for Issue #144. This module provides asset caching capabilities for improved performance including metadata caching, thumbnail caching, and cache management. """ import time from pathlib import Path from typing import Dict, Any, Optional, Tuple from dataclasses import dataclass, field from enum import Enum from collections import OrderedDict class CacheStrategy(Enum): """Cache eviction strategies.""" LRU = "lru" FIFO = "fifo" TTL = "ttl" @dataclass class CacheMetrics: """Cache performance metrics.""" total_requests: int = 0 cache_hits: int = 0 cache_misses: int = 0 evictions: int = 0 current_size_bytes: int = 0 @property def hit_rate(self) -> float: """Calculate cache hit rate.""" if self.total_requests == 0: return 0.0 return self.cache_hits / self.total_requests class AssetCache: """Asset caching system for metadata and thumbnails.""" def __init__(self, max_size_mb: int = 100, strategy: CacheStrategy = CacheStrategy.LRU, enable_metrics: bool = True): """Initialize asset cache.""" self.max_size_bytes = max_size_mb * 1024 * 1024 self.strategy = strategy self.enable_metrics = enable_metrics # Cache storage self._metadata_cache: OrderedDict = OrderedDict() self._thumbnail_cache: OrderedDict = OrderedDict() # Size tracking self.current_size_bytes = 0 # Metrics self._metrics = CacheMetrics() def store_metadata(self, content_hash: str, metadata: Dict[str, Any]): """Store asset metadata in cache.""" if self.enable_metrics: self._metrics.total_requests += 1 # Estimate size (simplified) estimated_size = len(str(metadata)) * 4 # Rough estimate # Check if we need to evict self._ensure_capacity(estimated_size) # Store metadata self._metadata_cache[content_hash] = { 'data': metadata, 'timestamp': time.time(), 'size': estimated_size } self.current_size_bytes += estimated_size if self.enable_metrics: self._metrics.cache_misses += 1 def get_metadata(self, content_hash: str) -> Optional[Dict[str, Any]]: """Retrieve asset metadata from cache.""" if self.enable_metrics: self._metrics.total_requests += 1 if content_hash in self._metadata_cache: # Move to end for LRU if self.strategy == CacheStrategy.LRU: metadata_entry = self._metadata_cache.pop(content_hash) self._metadata_cache[content_hash] = metadata_entry if self.enable_metrics: self._metrics.cache_hits += 1 return self._metadata_cache[content_hash]['data'] if self.enable_metrics: self._metrics.cache_misses += 1 return None def generate_and_cache_thumbnail(self, content_hash: str, image_path: Path, size: Tuple[int, int] = (150, 150)) -> bytes: """Generate and cache a thumbnail.""" thumbnail_key = f"{content_hash}_{size[0]}x{size[1]}" # Check if thumbnail already cached cached_thumbnail = self.get_thumbnail(content_hash, size) if cached_thumbnail: return cached_thumbnail # Generate thumbnail (simplified mock) thumbnail_data = f"thumbnail_{size[0]}x{size[1]}".encode() # Cache thumbnail estimated_size = len(thumbnail_data) self._ensure_capacity(estimated_size) self._thumbnail_cache[thumbnail_key] = { 'data': thumbnail_data, 'timestamp': time.time(), 'size': estimated_size } self.current_size_bytes += estimated_size return thumbnail_data def get_thumbnail(self, content_hash: str, size: Tuple[int, int]) -> Optional[bytes]: """Retrieve cached thumbnail.""" thumbnail_key = f"{content_hash}_{size[0]}x{size[1]}" if thumbnail_key in self._thumbnail_cache: # Move to end for LRU if self.strategy == CacheStrategy.LRU: thumbnail_entry = self._thumbnail_cache.pop(thumbnail_key) self._thumbnail_cache[thumbnail_key] = thumbnail_entry return self._thumbnail_cache[thumbnail_key]['data'] return None def invalidate(self, content_hash: str): """Invalidate cache entries for a specific asset.""" # Remove metadata if content_hash in self._metadata_cache: entry = self._metadata_cache.pop(content_hash) self.current_size_bytes -= entry['size'] # Remove thumbnails (find all sizes for this hash) keys_to_remove = [] for key in self._thumbnail_cache: if key.startswith(f"{content_hash}_"): keys_to_remove.append(key) for key in keys_to_remove: entry = self._thumbnail_cache.pop(key) self.current_size_bytes -= entry['size'] def get_hit_rate(self) -> float: """Get cache hit rate.""" return self._metrics.hit_rate def get_performance_metrics(self) -> Dict[str, Any]: """Get detailed performance metrics.""" return { 'total_requests': self._metrics.total_requests, 'cache_hits': self._metrics.cache_hits, 'cache_misses': self._metrics.cache_misses, 'hit_rate': self._metrics.hit_rate, 'evictions': self._metrics.evictions, 'current_size_bytes': self.current_size_bytes, 'max_size_bytes': self.max_size_bytes, 'size_utilization_percent': (self.current_size_bytes / self.max_size_bytes) * 100 } def _ensure_capacity(self, required_size: int): """Ensure cache has capacity for new entry.""" while (self.current_size_bytes + required_size) > self.max_size_bytes: if not self._metadata_cache and not self._thumbnail_cache: break # Cache is empty # Evict based on strategy if self.strategy == CacheStrategy.LRU: self._evict_lru() elif self.strategy == CacheStrategy.FIFO: self._evict_fifo() else: # TTL or default to LRU self._evict_lru() def _evict_lru(self): """Evict least recently used entry.""" # Find oldest entry across both caches oldest_metadata = None oldest_thumbnail = None if self._metadata_cache: oldest_metadata = next(iter(self._metadata_cache)) if self._thumbnail_cache: oldest_thumbnail = next(iter(self._thumbnail_cache)) # Compare timestamps if both exist metadata_entry = self._metadata_cache.get(oldest_metadata) if oldest_metadata else None thumbnail_entry = self._thumbnail_cache.get(oldest_thumbnail) if oldest_thumbnail else None if metadata_entry and thumbnail_entry: if metadata_entry['timestamp'] <= thumbnail_entry['timestamp']: self._evict_metadata_entry(oldest_metadata) else: self._evict_thumbnail_entry(oldest_thumbnail) elif metadata_entry: self._evict_metadata_entry(oldest_metadata) elif thumbnail_entry: self._evict_thumbnail_entry(oldest_thumbnail) def _evict_fifo(self): """Evict first in, first out entry.""" # For simplicity, just use LRU logic self._evict_lru() def _evict_metadata_entry(self, key: str): """Evict a metadata entry.""" if key in self._metadata_cache: entry = self._metadata_cache.pop(key) self.current_size_bytes -= entry['size'] if self.enable_metrics: self._metrics.evictions += 1 def _evict_thumbnail_entry(self, key: str): """Evict a thumbnail entry.""" if key in self._thumbnail_cache: entry = self._thumbnail_cache.pop(key) self.current_size_bytes -= entry['size'] if self.enable_metrics: self._metrics.evictions += 1 def clear(self): """Clear all cache entries.""" self._metadata_cache.clear() self._thumbnail_cache.clear() self.current_size_bytes = 0 self._metrics = CacheMetrics()