import time from abc import ABC, abstractmethod from typing import Any, Optional class CacheAdapter(ABC): """ Abstract base class for cache implementations. Allows custom cache backends (Redis, Memcached, etc.) to be plugged into the ApiClient for caching OIDC discovery metadata and JWKS. Example: class RedisCache(CacheAdapter): def __init__(self, redis_client): self.redis = redis_client def get(self, key: str) -> Optional[Any]: value = self.redis.get(key) return json.loads(value) if value else None def set(self, key: str, value: Any, ttl_seconds: Optional[int] = None) -> None: self.redis.set(key, json.dumps(value), ex=ttl_seconds) def delete(self, key: str) -> None: self.redis.delete(key) def clear(self) -> None: self.redis.flushdb() """ @abstractmethod def get(self, key: str) -> Optional[Any]: """ Get value from cache by key. Args: key: Cache key to retrieve Returns: Cached value if found and not expired, None otherwise """ pass @abstractmethod def set(self, key: str, value: Any, ttl_seconds: Optional[int] = None) -> None: """ Set value in cache with optional TTL. Args: key: Cache key to store value: Value to cache ttl_seconds: Time-to-live in seconds. None means no expiration. """ pass @abstractmethod def delete(self, key: str) -> None: """ Delete value from cache. Args: key: Cache key to delete """ pass @abstractmethod def clear(self) -> None: """Clear all cache entries.""" pass class InMemoryCache(CacheAdapter): """ Default in-memory cache implementation with LRU eviction. Designed for asyncio (single-threaded). For multi-threaded environments, implement a custom CacheAdapter with appropriate locking. Features: - TTL (time-to-live) support per entry using monotonic clock - LRU (Least Recently Used) eviction when max_entries reached - No external dependencies Args: max_entries: Maximum number of entries to cache. When exceeded, least recently used entry is evicted. Default: 100. Example: cache = InMemoryCache(max_entries=50) cache.set("key1", {"data": "value"}, ttl_seconds=600) value = cache.get("key1") # Returns {"data": "value"} """ def __init__(self, max_entries: int = 100): """ Initialize in-memory cache. Args: max_entries: Maximum number of cache entries (default: 100) """ self._cache: dict[str, tuple[Any, Optional[float]]] = {} self._max_entries = max_entries def get(self, key: str) -> Optional[Any]: """ Get value from cache by key. Updates access order for LRU tracking. Args: key: Cache key to retrieve Returns: Cached value if found and not expired, None otherwise """ if key not in self._cache: return None value, expiry = self._cache[key] if expiry is not None and time.monotonic() > expiry: del self._cache[key] return None del self._cache[key] self._cache[key] = (value, expiry) return value def set(self, key: str, value: Any, ttl_seconds: Optional[int] = None) -> None: """ Set value in cache with optional TTL. If cache is at max capacity, evicts least recently used entry. Args: key: Cache key to store value: Value to cache ttl_seconds: Time-to-live in seconds. None means no expiration. """ # If key exists, remove first so reinsert goes to end if key in self._cache: del self._cache[key] elif len(self._cache) >= self._max_entries: # Evict LRU: first key in dict is oldest oldest_key = next(iter(self._cache)) del self._cache[oldest_key] expiry = None if ttl_seconds is not None: expiry = time.monotonic() + ttl_seconds self._cache[key] = (value, expiry) def delete(self, key: str) -> None: """ Delete value from cache. Args: key: Cache key to delete """ self._cache.pop(key, None) def clear(self) -> None: """Clear all cache entries.""" self._cache.clear()