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

import json
from typing import Optional, Dict, Any

from utils import logger, RedisUtil
from .base_cache import BaseCache


class RedisCache(BaseCache):
    """基于Redis的缓存实现"""

    def __init__(self):
        self.redis = None

    async def initialize(self):
        """异步初始化方法，需在实例化后调用"""
        try:
            self.redis = await RedisUtil.get_redis_pool()
        except Exception as e:
            logger.error(f"Redis缓存初始化失败：{e}")
            raise

    async def get(self, cache_name: str, key: str) -> Optional[str]:
        """获取缓存值

        Args:
            cache_name: 缓存命名空间
            key: 缓存键
        Returns:
            缓存值（若存在）
        """
        return await self.redis.get(f"{cache_name}:{key}")

    async def set(self, cache_name: str, key: str, value: str) -> None:
        """设置缓存值

        Args:
            cache_name: 缓存命名空间
            key: 缓存键
            value: 缓存值
        """
        await self.redis.set(f"{cache_name}:{key}", value)

    async def get_dict(self, cache_name: str, key: str) -> Optional[Dict[str, Any]]:
        """获取字典类型缓存值
        Returns:
            Optional[Dict[str, Any]]: 返回指定的字典缓存，若不存在或解析失败则返回None
        """
        try:
            data = await self.redis.get(f"{cache_name}:{key}")
            if data:
                return json.loads(data)
            return None
        except json.JSONDecodeError:
            logger.error(f"缓存解析失败：{cache_name}:{key}")
            return None

    async def set_dict(self, cache_name: str, key: str, value: Dict[str, Any]) -> None:
        """设置字典类型缓存值
        Args:
            cache_name: 缓存命名空间
            key: 缓存键
            value: 字典类型缓存值
        """
        await self.redis.set(
            f"{cache_name}:{key}", json.dumps(value, ensure_ascii=False)
        )

    async def delete(self, cache_name: str, key: str) -> None:
        """删除缓存

        Args:
            cache_name: 缓存命名空间
            key: 缓存键
        """
        await self.redis.delete(f"{cache_name}:{key}")

    async def clear(self, cache_name: str, key: str = None) -> None:
        """清空指定缓存

        Args:
            cache_name: 缓存命名空间
            key:
        """
        if key:
            await self.redis.delete(f"{cache_name}:{key}")
        else:
            await self.redis.delete(f"{cache_name}:*")