diff --git a/zhenxun/services/cache/__init__.py b/zhenxun/services/cache/__init__.py index 4b98e871..7e6c5b1c 100644 --- a/zhenxun/services/cache/__init__.py +++ b/zhenxun/services/cache/__init__.py @@ -262,6 +262,8 @@ class CacheManager: _cache_backend: BaseCache | AioCache | None = None _registry: ClassVar[dict[str, CacheModel]] = {} _data: ClassVar[dict[str, CacheData]] = {} + _list_caches: ClassVar[dict[str, "CacheList"]] = {} + _dict_caches: ClassVar[dict[str, "CacheDict"]] = {} _enabled = False # 缓存启用标记 def __new__(cls) -> "CacheManager": @@ -290,6 +292,18 @@ class CacheManager: self.__class__._enabled = False logger.info("缓存功能已禁用", LOG_COMMAND) + def cache_dict(self, cache_type: str, expire: int = 0) -> CacheDict: + """获取缓存字典""" + if cache_type not in self._dict_caches: + self._dict_caches[cache_type] = CacheDict(cache_type, expire) + return self._dict_caches[cache_type] + + def cache_list(self, cache_type: str, expire: int = 0) -> CacheList: + """获取缓存列表""" + if cache_type not in self._list_caches: + self._list_caches[cache_type] = CacheList(cache_type, expire) + return self._list_caches[cache_type] + def listener(self, cache_type: str): """缓存监听器装饰器 diff --git a/zhenxun/services/cache/cache_containers.py b/zhenxun/services/cache/cache_containers.py index 91690e9a..b0efe3fb 100644 --- a/zhenxun/services/cache/cache_containers.py +++ b/zhenxun/services/cache/cache_containers.py @@ -1,12 +1,20 @@ -from typing import Any +from dataclasses import dataclass +import time +from typing import Any, Generic, TypeVar -from zhenxun.services.log import logger +T = TypeVar("T") -from .config import LOG_COMMAND + +@dataclass +class CacheData(Generic[T]): + """缓存数据类,存储数据和过期时间""" + + value: T + expire_time: float = 0 # 0表示永不过期 class CacheDict: - """全局缓存字典类,提供类似普通字典的接口,但数据可以在内存中共享""" + """缓存字典类,提供类似普通字典的接口,数据只存储在内存中""" def __init__(self, name: str, expire: int = 0): """初始化缓存字典 @@ -17,80 +25,7 @@ class CacheDict: """ self.name = name.upper() self.expire = expire - self._data = {} - # 自动尝试加载数据 - self._try_load() - - def _try_load(self): - """尝试加载数据(非异步)""" - try: - # 延迟导入,避免循环引用 - from zhenxun.services.cache import CacheRoot - - # 检查是否已有缓存数据 - if self.name in CacheRoot._data: - # 如果有,直接获取 - data = CacheRoot._data[self.name]._data - if isinstance(data, dict): - self._data = data - except Exception: - # 忽略错误,使用空字典 - pass - - async def load(self) -> bool: - """从缓存加载数据 - - 返回: - bool: 是否成功加载 - """ - try: - # 延迟导入,避免循环引用 - from zhenxun.services.cache import CacheRoot - - data = await CacheRoot.get_cache_data(self.name) - if isinstance(data, dict): - self._data = data - return True - return False - except Exception as e: - logger.error(f"加载缓存字典 {self.name} 失败", LOG_COMMAND, e=e) - return False - - async def save(self) -> bool: - """保存数据到缓存 - - 返回: - bool: 是否成功保存 - """ - try: - # 延迟导入,避免循环引用 - from zhenxun.services.cache import CacheData, CacheRoot - - # 检查缓存是否存在 - if self.name not in CacheRoot._data: - # 创建缓存 - async def get_func(): - return self._data - - CacheRoot._data[self.name] = CacheData( - name=self.name, - func=get_func, - expire=self.expire, - lazy_load=False, - cache=CacheRoot._cache, - ) - # 直接设置数据,避免调用func - CacheRoot._data[self.name]._data = self._data - else: - # 直接更新数据 - CacheRoot._data[self.name]._data = self._data - - # 保存数据 - await CacheRoot._data[self.name].set_data(self._data) - return True - except Exception as e: - logger.error(f"保存缓存字典 {self.name} 失败", LOG_COMMAND, e=e) - return False + self._data: dict[str, CacheData[Any]] = {} def __getitem__(self, key: str) -> Any: """获取字典项 @@ -101,7 +36,16 @@ class CacheDict: 返回: Any: 字典值 """ - return self._data.get(key) + data = self._data.get(key) + if data is None: + return None + + # 检查是否过期 + if data.expire_time > 0 and data.expire_time < time.time(): + del self._data[key] + return None + + return data.value def __setitem__(self, key: str, value: Any) -> None: """设置字典项 @@ -110,7 +54,12 @@ class CacheDict: key: 字典键 value: 字典值 """ - self._data[key] = value + # 计算过期时间 + expire_time = 0 + if self.expire > 0: + expire_time = time.time() + self.expire + + self._data[key] = CacheData(value=value, expire_time=expire_time) def __delitem__(self, key: str) -> None: """删除字典项 @@ -130,7 +79,16 @@ class CacheDict: 返回: bool: 是否存在 """ - return key in self._data + if key not in self._data: + return False + + # 检查是否过期 + data = self._data[key] + if data.expire_time > 0 and data.expire_time < time.time(): + del self._data[key] + return False + + return True def get(self, key: str, default: Any = None) -> Any: """获取字典项,如果不存在返回默认值 @@ -142,16 +100,25 @@ class CacheDict: 返回: Any: 字典值或默认值 """ - return self._data.get(key, default) + value = self[key] + return default if value is None else value - def set(self, key: str, value: Any) -> None: + def set(self, key: str, value: Any, expire: int | None = None) -> None: """设置字典项 参数: key: 字典键 value: 字典值 + expire: 过期时间(秒),为None时使用默认值 """ - self._data[key] = value + # 计算过期时间 + expire_time = 0 + if expire is not None and expire > 0: + expire_time = time.time() + expire + elif self.expire > 0: + expire_time = time.time() + self.expire + + self._data[key] = CacheData(value=value, expire_time=expire_time) def pop(self, key: str, default: Any = None) -> Any: """删除并返回字典项 @@ -163,7 +130,16 @@ class CacheDict: 返回: Any: 字典值或默认值 """ - return self._data.pop(key, default) + if key not in self._data: + return default + + data = self._data.pop(key) + + # 检查是否过期 + if data.expire_time > 0 and data.expire_time < time.time(): + return default + + return data.value def clear(self) -> None: """清空字典""" @@ -175,6 +151,8 @@ class CacheDict: 返回: list[str]: 键列表 """ + # 清理过期的键 + self._clean_expired() return list(self._data.keys()) def values(self) -> list[Any]: @@ -183,7 +161,9 @@ class CacheDict: 返回: list[Any]: 值列表 """ - return list(self._data.values()) + # 清理过期的键 + self._clean_expired() + return [data.value for data in self._data.values()] def items(self) -> list[tuple[str, Any]]: """获取所有键值对 @@ -191,7 +171,20 @@ class CacheDict: 返回: list[tuple[str, Any]]: 键值对列表 """ - return list(self._data.items()) + # 清理过期的键 + self._clean_expired() + return [(key, data.value) for key, data in self._data.items()] + + def _clean_expired(self) -> None: + """清理过期的键""" + now = time.time() + expired_keys = [ + key + for key, data in self._data.items() + if data.expire_time > 0 and data.expire_time < now + ] + for key in expired_keys: + del self._data[key] def __len__(self) -> int: """获取字典长度 @@ -199,6 +192,8 @@ class CacheDict: 返回: int: 字典长度 """ + # 清理过期的键 + self._clean_expired() return len(self._data) def __str__(self) -> str: @@ -207,11 +202,13 @@ class CacheDict: 返回: str: 字符串表示 """ + # 清理过期的键 + self._clean_expired() return f"CacheDict({self.name}, {len(self._data)} items)" class CacheList: - """全局缓存列表类,提供类似普通列表的接口,但数据可以在内存中共享""" + """缓存列表类,提供类似普通列表的接口,数据只存储在内存中""" def __init__(self, name: str, expire: int = 0): """初始化缓存列表 @@ -222,80 +219,12 @@ class CacheList: """ self.name = name.upper() self.expire = expire - self._data = [] - # 自动尝试加载数据 - self._try_load() + self._data: list[CacheData[Any]] = [] + self._expire_time = 0 - def _try_load(self): - """尝试加载数据(非异步)""" - try: - # 延迟导入,避免循环引用 - from zhenxun.services.cache import CacheRoot - - # 检查是否已有缓存数据 - if self.name in CacheRoot._data: - # 如果有,直接获取 - data = CacheRoot._data[self.name]._data - if isinstance(data, list): - self._data = data - except Exception: - # 忽略错误,使用空列表 - pass - - async def load(self) -> bool: - """从缓存加载数据 - - 返回: - bool: 是否成功加载 - """ - try: - # 延迟导入,避免循环引用 - from zhenxun.services.cache import CacheRoot - - data = await CacheRoot.get_cache_data(self.name) - if isinstance(data, list): - self._data = data - return True - return False - except Exception as e: - logger.error(f"加载缓存列表 {self.name} 失败", LOG_COMMAND, e=e) - return False - - async def save(self) -> bool: - """保存数据到缓存 - - 返回: - bool: 是否成功保存 - """ - try: - # 延迟导入,避免循环引用 - from zhenxun.services.cache import CacheData, CacheRoot - - # 检查缓存是否存在 - if self.name not in CacheRoot._data: - # 创建缓存 - async def get_func(): - return self._data - - CacheRoot._data[self.name] = CacheData( - name=self.name, - func=get_func, - expire=self.expire, - lazy_load=False, - cache=CacheRoot._cache, - ) - # 直接设置数据,避免调用func - CacheRoot._data[self.name]._data = self._data - else: - # 直接更新数据 - CacheRoot._data[self.name]._data = self._data - - # 保存数据 - await CacheRoot._data[self.name].set_data(self._data) - return True - except Exception as e: - logger.error(f"保存缓存列表 {self.name} 失败", LOG_COMMAND, e=e) - return False + # 如果设置了过期时间,计算整个列表的过期时间 + if self.expire > 0: + self._expire_time = time.time() + self.expire def __getitem__(self, index: int) -> Any: """获取列表项 @@ -306,8 +235,13 @@ class CacheList: 返回: Any: 列表值 """ + # 检查整个列表是否过期 + if self._is_expired(): + self.clear() + raise IndexError(f"列表索引 {index} 超出范围") + if 0 <= index < len(self._data): - return self._data[index] + return self._data[index].value raise IndexError(f"列表索引 {index} 超出范围") def __setitem__(self, index: int, value: Any) -> None: @@ -317,10 +251,17 @@ class CacheList: index: 列表索引 value: 列表值 """ + # 检查整个列表是否过期 + if self._is_expired(): + self.clear() + # 确保索引有效 while len(self._data) <= index: - self._data.append(None) - self._data[index] = value + self._data.append(CacheData(value=None)) + self._data[index] = CacheData(value=value) + + # 更新过期时间 + self._update_expire_time() def __delitem__(self, index: int) -> None: """删除列表项 @@ -328,8 +269,15 @@ class CacheList: 参数: index: 列表索引 """ + # 检查整个列表是否过期 + if self._is_expired(): + self.clear() + raise IndexError(f"列表索引 {index} 超出范围") + if 0 <= index < len(self._data): del self._data[index] + # 更新过期时间 + self._update_expire_time() else: raise IndexError(f"列表索引 {index} 超出范围") @@ -339,6 +287,9 @@ class CacheList: 返回: int: 列表长度 """ + # 检查整个列表是否过期 + if self._is_expired(): + self.clear() return len(self._data) def append(self, value: Any) -> None: @@ -347,7 +298,14 @@ class CacheList: 参数: value: 列表值 """ - self._data.append(value) + # 检查整个列表是否过期 + if self._is_expired(): + self.clear() + + self._data.append(CacheData(value=value)) + + # 更新过期时间 + self._update_expire_time() def extend(self, values: list[Any]) -> None: """扩展列表 @@ -355,7 +313,14 @@ class CacheList: 参数: values: 要添加的值列表 """ - self._data.extend(values) + # 检查整个列表是否过期 + if self._is_expired(): + self.clear() + + self._data.extend([CacheData(value=v) for v in values]) + + # 更新过期时间 + self._update_expire_time() def insert(self, index: int, value: Any) -> None: """插入列表项 @@ -364,7 +329,14 @@ class CacheList: index: 插入位置 value: 列表值 """ - self._data.insert(index, value) + # 检查整个列表是否过期 + if self._is_expired(): + self.clear() + + self._data.insert(index, CacheData(value=value)) + + # 更新过期时间 + self._update_expire_time() def pop(self, index: int = -1) -> Any: """删除并返回列表项 @@ -375,7 +347,20 @@ class CacheList: 返回: Any: 列表值 """ - return self._data.pop(index) + # 检查整个列表是否过期 + if self._is_expired(): + self.clear() + raise IndexError("从空列表中弹出") + + if not self._data: + raise IndexError("从空列表中弹出") + + item = self._data.pop(index) + + # 更新过期时间 + self._update_expire_time() + + return item.value def remove(self, value: Any) -> None: """删除第一个匹配的列表项 @@ -383,11 +368,26 @@ class CacheList: 参数: value: 要删除的值 """ - self._data.remove(value) + # 检查整个列表是否过期 + if self._is_expired(): + self.clear() + raise ValueError(f"{value} 不在列表中") + + # 查找匹配的项 + for i, item in enumerate(self._data): + if item.value == value: + del self._data[i] + # 更新过期时间 + self._update_expire_time() + return + + raise ValueError(f"{value} 不在列表中") def clear(self) -> None: """清空列表""" self._data.clear() + # 重置过期时间 + self._update_expire_time() def index(self, value: Any, start: int = 0, end: int | None = None) -> int: """查找值的索引 @@ -400,9 +400,18 @@ class CacheList: 返回: int: 索引位置 """ - return self._data.index( - value, start, end if end is not None else len(self._data) - ) + # 检查整个列表是否过期 + if self._is_expired(): + self.clear() + raise ValueError(f"{value} 不在列表中") + + end = end if end is not None else len(self._data) + + for i in range(start, min(end, len(self._data))): + if self._data[i].value == value: + return i + + raise ValueError(f"{value} 不在列表中") def count(self, value: Any) -> int: """计算值出现的次数 @@ -413,7 +422,23 @@ class CacheList: 返回: int: 出现次数 """ - return self._data.count(value) + # 检查整个列表是否过期 + if self._is_expired(): + self.clear() + return 0 + + return sum(1 for item in self._data if item.value == value) + + def _is_expired(self) -> bool: + """检查整个列表是否过期""" + return self._expire_time > 0 and self._expire_time < time.time() + + def _update_expire_time(self) -> None: + """更新过期时间""" + if self.expire > 0: + self._expire_time = time.time() + self.expire + else: + self._expire_time = 0 def __str__(self) -> str: """字符串表示 @@ -421,4 +446,7 @@ class CacheList: 返回: str: 字符串表示 """ + # 检查整个列表是否过期 + if self._is_expired(): + self.clear() return f"CacheList({self.name}, {len(self._data)} items)"