""" Caching module for Silver Table Assistant. Provides in-memory caching for performance optimization. """ import time import hashlib import json from typing import Any, Optional, Dict, Callable from functools import wraps import asyncio class Cache: """Simple in-memory cache with TTL support.""" def __init__(self, default_ttl: int = 3600): # 1 hour default TTL self._cache: Dict[str, Dict[str, Any]] = {} self.default_ttl = default_ttl def _generate_key(self, prefix: str, *args, **kwargs) -> str: """Generate a cache key from function arguments.""" # Convert args and kwargs to a sorted string representation key_data = { "args": args, "kwargs": kwargs } key_string = json.dumps(key_data, sort_keys=True, default=str) key_hash = hashlib.md5(key_string.encode()).hexdigest() return f"{prefix}:{key_hash}" def get(self, key: str) -> Optional[Any]: """Get value from cache if not expired.""" if key in self._cache: cache_entry = self._cache[key] if time.time() < cache_entry["expires_at"]: return cache_entry["value"] else: # Remove expired entry del self._cache[key] return None def set(self, key: str, value: Any, ttl: Optional[int] = None) -> None: """Set value in cache with TTL.""" ttl = ttl or self.default_ttl self._cache[key] = { "value": value, "expires_at": time.time() + ttl } def delete(self, key: str) -> None: """Delete key from cache.""" if key in self._cache: del self._cache[key] def clear(self) -> None: """Clear all cache entries.""" self._cache.clear() def cleanup_expired(self) -> int: """Remove all expired entries and return count removed.""" current_time = time.time() 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] return len(expired_keys) def get_stats(self) -> Dict[str, Any]: """Get cache statistics.""" current_time = time.time() total_entries = len(self._cache) expired_entries = sum( 1 for entry in self._cache.values() if current_time >= entry["expires_at"] ) return { "total_entries": total_entries, "active_entries": total_entries - expired_entries, "expired_entries": expired_entries, "cache_hit_potential": "high" if total_entries > 100 else "medium" if total_entries > 10 else "low" } # Global cache instances document_cache = Cache(default_ttl=1800) # 30 minutes for document queries nutrition_cache = Cache(default_ttl=3600) # 1 hour for nutrition calculations user_context_cache = Cache(default_ttl=900) # 15 minutes for user context def cache_result(cache_instance: Cache, prefix: str, ttl: Optional[int] = None): """Decorator to cache function results.""" def decorator(func: Callable): @wraps(func) async def async_wrapper(*args, **kwargs): # Generate cache key cache_key = cache_instance._generate_key(prefix, *args, **kwargs) # Try to get from cache cached_result = cache_instance.get(cache_key) if cached_result is not None: return cached_result # Execute function and cache result if asyncio.iscoroutinefunction(func): result = await func(*args, **kwargs) else: result = func(*args, **kwargs) cache_instance.set(cache_key, result, ttl) return result @wraps(func) def sync_wrapper(*args, **kwargs): # Generate cache key cache_key = cache_instance._generate_key(prefix, *args, **kwargs) # Try to get from cache cached_result = cache_instance.get(cache_key) if cached_result is not None: return cached_result # Execute function and cache result result = func(*args, **kwargs) cache_instance.set(cache_key, result, ttl) return result # Choose wrapper based on function type if asyncio.iscoroutinefunction(func): return async_wrapper else: return sync_wrapper return decorator class NutritionCache: """Specialized cache for nutrition calculations.""" @staticmethod def get_menu_item_nutrition(menu_item_id: int) -> Optional[Dict[str, Any]]: """Get cached nutrition data for a menu item.""" key = f"nutrition:menu_item:{menu_item_id}" return nutrition_cache.get(key) @staticmethod def set_menu_item_nutrition(menu_item_id: int, nutrition_data: Dict[str, Any], ttl: Optional[int] = None) -> None: """Cache nutrition data for a menu item.""" key = f"nutrition:menu_item:{menu_item_id}" nutrition_cache.set(key, nutrition_data, ttl) @staticmethod def get_user_nutrition_summary(user_id: str, days: int = 7) -> Optional[Dict[str, Any]]: """Get cached nutrition summary for a user.""" key = f"nutrition:summary:{user_id}:{days}" return nutrition_cache.get(key) @staticmethod def set_user_nutrition_summary(user_id: str, days: int, summary_data: Dict[str, Any], ttl: Optional[int] = None) -> None: """Cache nutrition summary for a user.""" key = f"nutrition:summary:{user_id}:{days}" nutrition_cache.set(key, summary_data, ttl) @staticmethod def invalidate_user_nutrition(user_id: str) -> None: """Invalidate all nutrition cache for a user.""" # Remove all nutrition-related entries for this user keys_to_remove = [ key for key in nutrition_cache._cache.keys() if key.startswith(f"nutrition:summary:{user_id}") ] for key in keys_to_remove: nutrition_cache.delete(key) class DocumentCache: """Specialized cache for document queries.""" @staticmethod def get_relevant_documents(query: str, k: int, score_threshold: Optional[float] = None) -> Optional[list]: """Get cached document search results.""" threshold_key = f":{score_threshold}" if score_threshold else "" key = f"documents:query:{hashlib.md5(query.encode()).hexdigest()}:k{k}{threshold_key}" return document_cache.get(key) @staticmethod def set_relevant_documents(query: str, k: int, documents: list, score_threshold: Optional[float] = None, ttl: Optional[int] = None) -> None: """Cache document search results.""" threshold_key = f":{score_threshold}" if score_threshold else "" key = f"documents:query:{hashlib.md5(query.encode()).hexdigest()}:k{k}{threshold_key}" document_cache.set(key, documents, ttl) @staticmethod def invalidate_document_cache() -> None: """Invalidate all document cache entries.""" document_cache.clear() class UserContextCache: """Specialized cache for user context data.""" @staticmethod def get_user_context(user_id: str) -> Optional[Dict[str, Any]]: """Get cached user context.""" key = f"context:user:{user_id}" return user_context_cache.get(key) @staticmethod def set_user_context(user_id: str, context_data: Dict[str, Any], ttl: Optional[int] = None) -> None: """Cache user context data.""" key = f"context:user:{user_id}" user_context_cache.set(key, context_data, ttl) @staticmethod def invalidate_user_context(user_id: str) -> None: """Invalidate user context cache.""" key = f"context:user:{user_id}" user_context_cache.delete(key) # Cache management utilities def get_cache_stats() -> Dict[str, Any]: """Get statistics for all caches.""" return { "document_cache": document_cache.get_stats(), "nutrition_cache": nutrition_cache.get_stats(), "user_context_cache": user_context_cache.get_stats() } def cleanup_all_caches() -> Dict[str, int]: """Clean up expired entries from all caches.""" return { "document_cache": document_cache.cleanup_expired(), "nutrition_cache": nutrition_cache.cleanup_expired(), "user_context_cache": user_context_cache.cleanup_expired() } def invalidate_user_cache(user_id: str) -> None: """Invalidate all cache entries for a specific user.""" NutritionCache.invalidate_user_nutrition(user_id) UserContextCache.invalidate_user_context(user_id)