File size: 2,087 Bytes
b4f2238 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 | """配置缓存模块"""
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)
|