""" cache_manager.py — Cache Layer Distribuito su Supabase A Architettura: - Supabase A funge da archivio cache distribuito (read-only, non-critical) - TTL configurabile per invalidazione automatica - Fallback a B se A non disponibile - Supporta cache per query, memoria, embeddings, conversazioni Strategie di Caching: 1. Query Cache: Risultati query SELECT memorizzati con TTL 2. Memory Cache: Memoria agente memorizzata per accesso veloce 3. Embedding Cache: Embeddings pre-calcolati per RAG 4. Conversation Cache: Conversazioni recenti per accesso veloce Invalidazione: - TTL-based: Scadenza automatica dopo TTL - Event-based: Invalidazione su INSERT/UPDATE/DELETE su B - Manual: Invalidazione esplicita via API Statistiche: - Hit rate: % di richieste servite da cache - Miss rate: % di richieste non in cache - Eviction rate: % di entry rimosse per TTL """ import os import asyncio import logging import hashlib import json from typing import Optional, Dict, List, Any, Callable from datetime import datetime, timedelta from enum import Enum import time _logger = logging.getLogger("cache_manager") # ── Configurazione Cache ─────────────────────────────────────────────────── CACHE_ENABLED = os.getenv("SUPABASE_CACHE_ENABLED", "true").lower() == "true" CACHE_TTL_DEFAULT = int(os.getenv("SUPABASE_CACHE_TTL", "3600")) # 1 ora CACHE_MAX_SIZE = int(os.getenv("SUPABASE_CACHE_MAX_SIZE", "10000")) # max entries CACHE_STATS_ENABLED = os.getenv("SUPABASE_CACHE_STATS_ENABLED", "true").lower() == "true" # ── Enumerazione Strategie Cache ─────────────────────────────────────────── class CacheStrategy(str, Enum): QUERY = "query" # Cache risultati query MEMORY = "memory" # Cache memoria agente EMBEDDING = "embedding" # Cache embeddings CONVERSATION = "conversation" # Cache conversazioni ANALYTICS = "analytics" # Cache analytics/reporting # ── TTL per Strategia ───────────────────────────────────────────────────── CACHE_TTL_BY_STRATEGY = { CacheStrategy.QUERY: int(os.getenv("CACHE_TTL_QUERY", "600")), # 10 min CacheStrategy.MEMORY: int(os.getenv("CACHE_TTL_MEMORY", "1800")), # 30 min CacheStrategy.EMBEDDING: int(os.getenv("CACHE_TTL_EMBEDDING", "7200")), # 2 ore CacheStrategy.CONVERSATION: int(os.getenv("CACHE_TTL_CONVERSATION", "3600")), # 1 ora CacheStrategy.ANALYTICS: int(os.getenv("CACHE_TTL_ANALYTICS", "300")), # 5 min } # ── Configurazione Supabase A ────────────────────────────────────────────── SUPABASE_A_URL = os.getenv("SUPABASE_URL_A", "") SUPABASE_A_KEY = os.getenv("SUPABASE_KEY_A", "") SUPABASE_B_URL = os.getenv("SUPABASE_URL", "") SUPABASE_B_KEY = os.getenv("SUPABASE_KEY", "") # ── Client Cache ─────────────────────────────────────────────────────────── class CacheClient: """Client per gestire cache su Supabase A.""" def __init__(self, url: str, key: str): self.url = url self.key = key self.base_url = f"{url}/rest/v1" if url else None self._stats = { "hits": 0, "misses": 0, "sets": 0, "deletes": 0, "evictions": 0, } def _generate_cache_key(self, strategy: str, identifier: str) -> str: """Genera una chiave cache univoca.""" combined = f"{strategy}:{identifier}" return hashlib.sha256(combined.encode()).hexdigest()[:32] async def get(self, strategy: CacheStrategy, identifier: str) -> Optional[Dict]: """Recupera un valore dalla cache.""" if not CACHE_ENABLED or not self.base_url: return None cache_key = self._generate_cache_key(strategy.value, identifier) try: import httpx url = f"{self.base_url}/cache_entries?cache_key=eq.{cache_key}" headers = { "apikey": self.key, "Authorization": f"Bearer {self.key}", "Content-Type": "application/json", } async with httpx.AsyncClient() as client: response = await client.get(url, headers=headers, timeout=5.0) if response.status_code == 200: rows = response.json() if rows: entry = rows[0] # Verifica TTL if self._is_expired(entry): # Elimina entry scaduta await self._delete_entry(cache_key) if CACHE_STATS_ENABLED: self._stats["evictions"] += 1 return None # Hit if CACHE_STATS_ENABLED: self._stats["hits"] += 1 return { "value": entry.get("value"), "cached_at": entry.get("created_at"), "ttl_remaining": self._get_ttl_remaining(entry), } # Miss if CACHE_STATS_ENABLED: self._stats["misses"] += 1 return None except Exception as exc: _logger.error(f"Cache get error: {exc}") return None async def set( self, strategy: CacheStrategy, identifier: str, value: Any, ttl: Optional[int] = None, ) -> bool: """Memorizza un valore nella cache.""" if not CACHE_ENABLED or not self.base_url: return False cache_key = self._generate_cache_key(strategy.value, identifier) ttl = ttl or CACHE_TTL_BY_STRATEGY.get(strategy, CACHE_TTL_DEFAULT) try: import httpx url = f"{self.base_url}/cache_entries" headers = { "apikey": self.key, "Authorization": f"Bearer {self.key}", "Content-Type": "application/json", } data = { "cache_key": cache_key, "strategy": strategy.value, "identifier": identifier, "value": json.dumps(value) if not isinstance(value, str) else value, "ttl_seconds": ttl, "created_at": datetime.now().isoformat(), "expires_at": (datetime.now() + timedelta(seconds=ttl)).isoformat(), } async with httpx.AsyncClient() as client: response = await client.post(url, json=data, headers=headers, timeout=5.0) if response.status_code in (200, 201): if CACHE_STATS_ENABLED: self._stats["sets"] += 1 return True else: _logger.warning(f"Cache set failed: {response.status_code}") return False except Exception as exc: _logger.error(f"Cache set error: {exc}") return False async def delete(self, strategy: CacheStrategy, identifier: str) -> bool: """Elimina un valore dalla cache.""" if not CACHE_ENABLED or not self.base_url: return False cache_key = self._generate_cache_key(strategy.value, identifier) return await self._delete_entry(cache_key) async def _delete_entry(self, cache_key: str) -> bool: """Elimina un entry dalla cache per chiave.""" try: import httpx url = f"{self.base_url}/cache_entries?cache_key=eq.{cache_key}" headers = { "apikey": self.key, "Authorization": f"Bearer {self.key}", } async with httpx.AsyncClient() as client: response = await client.delete(url, headers=headers, timeout=5.0) if response.status_code in (200, 204): if CACHE_STATS_ENABLED: self._stats["deletes"] += 1 return True return False except Exception as exc: _logger.error(f"Cache delete error: {exc}") return False def _is_expired(self, entry: Dict) -> bool: """Verifica se un entry è scaduto.""" expires_at = entry.get("expires_at") if not expires_at: return False try: expires_dt = datetime.fromisoformat(expires_at) return datetime.now() > expires_dt except: return False def _get_ttl_remaining(self, entry: Dict) -> int: """Calcola il TTL rimanente in secondi.""" expires_at = entry.get("expires_at") if not expires_at: return 0 try: expires_dt = datetime.fromisoformat(expires_at) remaining = (expires_dt - datetime.now()).total_seconds() return max(0, int(remaining)) except: return 0 def get_stats(self) -> Dict[str, Any]: """Ritorna statistiche cache.""" total = self._stats["hits"] + self._stats["misses"] hit_rate = (self._stats["hits"] / total * 100) if total > 0 else 0 return { "enabled": CACHE_ENABLED, "hits": self._stats["hits"], "misses": self._stats["misses"], "sets": self._stats["sets"], "deletes": self._stats["deletes"], "evictions": self._stats["evictions"], "hit_rate_percent": round(hit_rate, 2), "total_requests": total, } def reset_stats(self): """Resetta le statistiche.""" self._stats = { "hits": 0, "misses": 0, "sets": 0, "deletes": 0, "evictions": 0, } # ── Cache Wrapper con Fallback ───────────────────────────────────────────── class CacheManager: """Gestore cache con fallback intelligente.""" def __init__(self): self.cache_client = CacheClient(SUPABASE_A_URL, SUPABASE_A_KEY) if SUPABASE_A_URL else None self.fallback_client = CacheClient(SUPABASE_B_URL, SUPABASE_B_KEY) if SUPABASE_B_URL else None async def get_or_fetch( self, strategy: CacheStrategy, identifier: str, fetch_fn: Callable, ttl: Optional[int] = None, ) -> Any: """ Recupera valore da cache o lo genera con fetch_fn. Logica: 1. Prova a leggere da cache A 2. Se miss, chiama fetch_fn 3. Memorizza risultato in cache A 4. Ritorna valore """ # Prova cache A if self.cache_client: cached = await self.cache_client.get(strategy, identifier) if cached: _logger.debug(f"Cache hit: {strategy.value}:{identifier}") return json.loads(cached["value"]) if isinstance(cached["value"], str) else cached["value"] # Cache miss → fetch _logger.debug(f"Cache miss: {strategy.value}:{identifier}, fetching...") try: value = await fetch_fn() if asyncio.iscoroutinefunction(fetch_fn) else fetch_fn() except Exception as exc: _logger.error(f"Fetch error: {exc}") return None # Memorizza in cache A if self.cache_client and value is not None: ttl = ttl or CACHE_TTL_BY_STRATEGY.get(strategy, CACHE_TTL_DEFAULT) await self.cache_client.set(strategy, identifier, value, ttl) return value async def invalidate(self, strategy: CacheStrategy, identifier: str) -> bool: """Invalida un entry cache.""" if self.cache_client: return await self.cache_client.delete(strategy, identifier) return False async def invalidate_pattern(self, strategy: CacheStrategy, pattern: str) -> int: """Invalida tutti gli entry che corrispondono a un pattern.""" # TODO: Implementare pattern matching return 0 def get_stats(self) -> Dict[str, Any]: """Ritorna statistiche cache.""" if self.cache_client: return self.cache_client.get_stats() return {"enabled": False} def reset_stats(self): """Resetta statistiche.""" if self.cache_client: self.cache_client.reset_stats() # ── Istanza Globale ─────────────────────────────────────────────────────── _cache_manager = CacheManager() # ── Funzioni Pubbliche ──────────────────────────────────────────────────── async def get_cached( strategy: CacheStrategy, identifier: str, fetch_fn: Callable, ttl: Optional[int] = None, ) -> Any: """Recupera valore da cache o lo genera.""" return await _cache_manager.get_or_fetch(strategy, identifier, fetch_fn, ttl) async def invalidate_cache(strategy: CacheStrategy, identifier: str) -> bool: """Invalida un entry cache.""" return await _cache_manager.invalidate(strategy, identifier) def get_cache_stats() -> Dict[str, Any]: """Ritorna statistiche cache.""" return _cache_manager.get_stats() def reset_cache_stats(): """Resetta statistiche cache.""" _cache_manager.reset_stats() # ── Decorator per Caching Automatico ─────────────────────────────────────── def cached(strategy: CacheStrategy, ttl: Optional[int] = None): """ Decorator per caching automatico di funzioni. Uso: @cached(CacheStrategy.QUERY, ttl=600) async def get_user_data(user_id: str): return await fetch_user_from_db(user_id) """ def decorator(func: Callable): async def wrapper(*args, **kwargs): # Genera identifier da args/kwargs identifier = f"{func.__name__}:{str(args)}:{str(kwargs)}" # Usa cache manager return await _cache_manager.get_or_fetch( strategy, identifier, lambda: func(*args, **kwargs), ttl, ) return wrapper return decorator