✨ feat(cache): 增强缓存管理，新增缓存字典和缓存列表功能，支持过期时间管理

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):
         """缓存监听器装饰器
 

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._data: dict[str, CacheData[Any]] = {}
-        # 自动尝试加载数据
-        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
 
     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._data: list[CacheData[Any]] = []
-        # 自动尝试加载数据
+        self._expire_time = 0
-        self._try_load()
 
-    def _try_load(self):
+        # 如果设置了过期时间，计算整个列表的过期时间
-        """尝试加载数据（非异步）"""
+        if self.expire > 0:
-        try:
+            self._expire_time = time.time() + self.expire
-            # 延迟导入，避免循环引用
-            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
 
     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.append(CacheData(value=None))
-        self._data[index] = value
+        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)"