import time from typing import OrderedDict class CacheEntry: def __init__(self, value, ttl): self.value = value self.expiry = time.time() + ttl class MultiDomainCache: def __init__(self, max_entries=128, ttl=300): self.max_entries = max_entries self.ttl = ttl self.cache = {} def _evict(self, domain): """Evict the oldest entries in the domain cache until max_entries is not exceeded.""" domain_cache = self.cache[domain] while len(domain_cache) > self.max_entries: domain_cache.popitem(last=False) def _is_expired(self, entry): """Check if a cache entry is expired.""" return entry.expiry < time.time() def add(self, domain, key, value, ttl=None): """Add an entry to the cache.""" ttl = ttl if ttl is not None else self.ttl if domain not in self.cache: self.cache[domain] = OrderedDict() self.cache[domain][key] = CacheEntry(value, ttl) self._evict(domain) def get(self, domain, key): """Get an entry from the cache. Return (False, None) if not found or expired.""" if domain in self.cache and key in self.cache[domain]: entry = self.cache[domain][key] if not self._is_expired(entry): return True, entry.value else: # Remove expired entry del self.cache[domain][key] # Remove domain cache if empty if not self.cache[domain]: del self.cache[domain] return False, None def invalidate(self, domain, key_prefix): """Invalidate (remove) entries from the cache that match the key prefix.""" string_prefix = str(key_prefix) if not isinstance(key_prefix, str) else key_prefix if domain in self.cache: domain_cache = self.cache[domain] # Collect keys to delete to avoid modifying dict during iteration keys_to_delete = [key for key in domain_cache if str(key).startswith(string_prefix)] for key in keys_to_delete: del domain_cache[key] # Remove domain cache if empty if not domain_cache: del self.cache[domain] def update(self, domain, key, value, ttl=None): """Update an entry in the cache.""" ttl = ttl if ttl is not None else self.ttl if domain in self.cache and key in self.cache[domain]: self.cache[domain][key] = CacheEntry(value, ttl) self._evict(domain) else: self.add(domain, key, value, ttl) # Example usage multidomain_cache = MultiDomainCache(ttl=60*60*5)