VberAdminPy/VapAdmin/BACKEND/core/cache/memory_cache.py

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]