张翔
146 lines
3.3 KiB
Python
146 lines
3.3 KiB
Python
"""
|
|
缓存模块
|
|
|
|
提供内存缓存功能。
|
|
"""
|
|
|
|
import time
|
|
import threading
|
|
from typing import Dict, Any, Optional
|
|
from dataclasses import dataclass, field
|
|
|
|
|
|
@dataclass
|
|
class CacheEntry:
|
|
"""缓存条目"""
|
|
value: Any
|
|
expires_at: Optional[float] = None
|
|
created_at: float = field(default_factory=time.time)
|
|
|
|
|
|
class Cache:
|
|
"""缓存类"""
|
|
|
|
def __init__(self):
|
|
self._data: Dict[str, CacheEntry] = {}
|
|
self._lock = threading.Lock()
|
|
|
|
def set(self, key: str, value: Any, ttl: Optional[int] = None) -> None:
|
|
"""
|
|
设置缓存
|
|
|
|
Args:
|
|
key: 缓存键
|
|
value: 缓存值
|
|
ttl: 过期时间(秒)
|
|
"""
|
|
with self._lock:
|
|
expires_at = None
|
|
if ttl is not None:
|
|
expires_at = time.time() + ttl
|
|
|
|
self._data[key] = CacheEntry(
|
|
value=value,
|
|
expires_at=expires_at
|
|
)
|
|
|
|
def get(self, key: str) -> Optional[Any]:
|
|
"""
|
|
获取缓存
|
|
|
|
Args:
|
|
key: 缓存键
|
|
|
|
Returns:
|
|
缓存值，如果不存在或已过期则返回None
|
|
"""
|
|
with self._lock:
|
|
entry = self._data.get(key)
|
|
if entry is None:
|
|
return None
|
|
|
|
# 检查是否过期
|
|
if entry.expires_at is not None and time.time() > entry.expires_at:
|
|
del self._data[key]
|
|
return None
|
|
|
|
return entry.value
|
|
|
|
def delete(self, key: str) -> bool:
|
|
"""
|
|
删除缓存
|
|
|
|
Args:
|
|
key: 缓存键
|
|
|
|
Returns:
|
|
是否成功删除
|
|
"""
|
|
with self._lock:
|
|
if key in self._data:
|
|
del self._data[key]
|
|
return True
|
|
return False
|
|
|
|
def clear(self) -> None:
|
|
"""清除所有缓存"""
|
|
with self._lock:
|
|
self._data.clear()
|
|
|
|
def get_stats(self) -> Dict[str, Any]:
|
|
"""
|
|
获取缓存统计信息
|
|
|
|
Returns:
|
|
统计信息字典
|
|
"""
|
|
with self._lock:
|
|
# 清理过期条目
|
|
current_time = time.time()
|
|
expired_keys = [
|
|
key for key, entry in self._data.items()
|
|
if entry.expires_at is not None and current_time > entry.expires_at
|
|
]
|
|
for key in expired_keys:
|
|
del self._data[key]
|
|
|
|
return {
|
|
"size": len(self._data),
|
|
"keys": list(self._data.keys())
|
|
}
|
|
|
|
def has(self, key: str) -> bool:
|
|
"""
|
|
检查缓存是否存在
|
|
|
|
Args:
|
|
key: 缓存键
|
|
|
|
Returns:
|
|
是否存在
|
|
"""
|
|
return self.get(key) is not None
|
|
|
|
def get_all(self) -> Dict[str, Any]:
|
|
"""
|
|
获取所有缓存
|
|
|
|
Returns:
|
|
所有缓存数据
|
|
"""
|
|
with self._lock:
|
|
result = {}
|
|
current_time = time.time()
|
|
|
|
for key, entry in self._data.items():
|
|
# 检查是否过期
|
|
if entry.expires_at is not None and current_time > entry.expires_at:
|
|
continue
|
|
result[key] = entry.value
|
|
|
|
return result
|
|
|
|
|
|
# 全局缓存实例
|
|
cache = Cache()
|