from typing import Optional, Dict, Any from .base_cache import BaseCache class MemoryCache(BaseCache): """基于内存的缓存实现""" def __init__(self): self._cache = {} # 现在存储结构为 {cache_name: {key: value}} async def get(self, cache_name: str, key: str) -> Optional[str]: """获取缓存值""" return self._cache.get(cache_name, {}).get(key) async def set(self, cache_name: str, key: str, value: str) -> None: """设置缓存值""" if cache_name not in self._cache: self._cache[cache_name] = {} self._cache[cache_name][key] = value async def get_dict(self, cache_name: str, key: str) -> Optional[Dict[str, Any]]: """获取字典类型缓存值 Returns: Optional[Dict[str, Any]]: 返回指定的字典缓存,若不存在则返回None """ value = self._cache.get(cache_name, {}).get(key) if isinstance(value, dict): return value return None async def set_dict(self, cache_name: str, key: str, value: Dict[str, Any]) -> None: """设置字典类型缓存值 Args: cache_name: 缓存命名空间 key: 缓存键 value: 字典类型缓存值 """ if cache_name not in self._cache: self._cache[cache_name] = None self._cache[cache_name][key] = value async def delete(self, cache_name: str, key: str) -> None: """删除缓存""" if cache_name in self._cache and key in self._cache[cache_name]: del self._cache[cache_name][key] async def clear(self, cache_name: str, key: str = None) -> None: """清空指定缓存""" if cache_name in self._cache: if key: del self._cache[cache_name][key] else: del self._cache[cache_name]