Spaces:
Runtime error
Runtime error
| 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) | |