| """配置缓存模块""" |
|
|
| import threading |
| import time |
| from dataclasses import dataclass, field |
| from typing import Any |
|
|
|
|
| @dataclass |
| class CacheEntry: |
| value: Any |
| created_at: float = field(default_factory=time.time) |
| last_access: float = field(default_factory=time.time) |
| access_count: int = 0 |
|
|
| def touch(self) -> None: |
| self.last_access = time.time() |
| self.access_count += 1 |
|
|
| def is_expired(self, ttl: float) -> bool: |
| return ttl > 0 and (time.time() - self.created_at) > ttl |
|
|
|
|
| class ConfigCache: |
| def __init__(self, max_size: int = 1000, ttl: float = 3600.0) -> None: |
| self.max_size = max_size |
| self.ttl = ttl |
| self._cache: dict[str, CacheEntry] = {} |
| self._lock = threading.RLock() |
|
|
| def get(self, key: str) -> Any | None: |
| with self._lock: |
| entry = self._cache.get(key) |
| if entry is None: |
| return None |
| if entry.is_expired(self.ttl): |
| del self._cache[key] |
| return None |
| entry.touch() |
| return entry.value |
|
|
| def set(self, key: str, value: Any) -> None: |
| with self._lock: |
| if len(self._cache) >= self.max_size and key not in self._cache: |
| self._evict_lru() |
| self._cache[key] = CacheEntry(value) |
|
|
| def delete(self, key: str) -> bool: |
| with self._lock: |
| if key in self._cache: |
| del self._cache[key] |
| return True |
| return False |
|
|
| def clear(self) -> None: |
| with self._lock: |
| self._cache.clear() |
|
|
| def _evict_lru(self) -> None: |
| if not self._cache: |
| return |
| lru_key = min(self._cache, key=lambda k: self._cache[k].last_access) |
| del self._cache[lru_key] |
|
|
| def cleanup_expired(self) -> int: |
| if self.ttl <= 0: |
| return 0 |
| with self._lock: |
| expired = [k for k, e in self._cache.items() if e.is_expired(self.ttl)] |
| for k in expired: |
| del self._cache[k] |
| return len(expired) |
|
|
