import time from typing import Dict, Any, Optional, Callable, TypeVar, Generic import logging logger = logging.getLogger(__name__) T = TypeVar('T') class Cache(Generic[T]): """ A simple in-memory cache implementation with TTL (time to live). """ def __init__(self, ttl: int = 300): """ Initialize the cache. Args: ttl: Time to live for cache entries in seconds. Default is 300 seconds (5 minutes). """ self._cache: Dict[str, Dict[str, Any]] = {} self.ttl = ttl def get(self, key: str) -> Optional[T]: """ Get a value from the cache. Args: key: The cache key. Returns: The cached value if it exists and hasn't expired, None otherwise. """ if key in self._cache: entry = self._cache[key] if entry["expires"] > time.time(): logger.debug(f"Cache hit for key: {key}") return entry["value"] else: # Remove expired entry logger.debug(f"Cache expired for key: {key}") del self._cache[key] return None def set(self, key: str, value: T, ttl: Optional[int] = None) -> None: """ Set a value in the cache. Args: key: The cache key. value: The value to cache. ttl: Optional custom TTL for this entry. If not provided, the default TTL is used. """ expiry = time.time() + (ttl if ttl is not None else self.ttl) self._cache[key] = {"value": value, "expires": expiry} logger.debug(f"Cache set for key: {key}") def delete(self, key: str) -> None: """ Delete a value from the cache. Args: key: The cache key. """ if key in self._cache: del self._cache[key] logger.debug(f"Cache deleted for key: {key}") def clear(self) -> None: """ Clear all entries from the cache. """ self._cache.clear() logger.debug("Cache cleared") def get_or_set(self, key: str, value_func: Callable[[], T], ttl: Optional[int] = None) -> T: """ Get a value from the cache, or set it if it doesn't exist. Args: key: The cache key. value_func: A function that returns the value to cache if it doesn't exist. ttl: Optional custom TTL for this entry. If not provided, the default TTL is used. Returns: The cached value. """ value = self.get(key) if value is None: value = value_func() self.set(key, value, ttl) return value # Create global cache instance cache = Cache(ttl=300) # 5 minutes TTL api_cache = Cache(ttl=60) # 1 minute TTL for API responses