RuoYi-Vue3-FastAPI/ruoyi-fastapi-backend/cli/runtime/cache/service.py

from typing import Any

from cli.exit_codes import RUNTIME_ERROR

from .gateway import REDIS_TTL_KEY_NOT_FOUND, REDIS_TTL_PERSISTENT, CacheInfrastructureGateway
from .support import CacheDomainSupport, CacheRedisSupport


class CacheRuntimeService:
    """
    缓存运行时服务。

    该服务作为缓存运行时 facade，对外统一暴露缓存统计、键列表、键值、
    TTL 读取，以及缓存清理和预热入口。

    :param infrastructure_gateway: 缓存基础设施网关
    :param domain_support: 缓存领域支持对象
    :param redis_support: 缓存 Redis 访问支持对象
    """

    def __init__(
        self,
        *,
        infrastructure_gateway: CacheInfrastructureGateway | None = None,
        domain_support: CacheDomainSupport | None = None,
        redis_support: CacheRedisSupport | None = None,
    ) -> None:
        """
        初始化缓存运行时服务。

        :param infrastructure_gateway: 缓存基础设施网关
        :param domain_support: 缓存领域支持对象
        :param redis_support: 缓存 Redis 访问支持对象
        :return: None
        """
        self.infrastructure_gateway = infrastructure_gateway or CacheInfrastructureGateway()
        self.domain_support = domain_support or CacheDomainSupport(self.infrastructure_gateway)
        self.redis_support = redis_support or CacheRedisSupport(self.infrastructure_gateway, self.domain_support)

    async def get_cache_stats(self) -> dict[str, Any]:
        """
        获取缓存统计信息。

        :return: 缓存统计结果
        """
        redis_error = self.infrastructure_gateway.get_redis_error_class()
        try:
            async with self.redis_support.redis_session() as (redis, _redis_util):
                info = await redis.info()
                db_size = await redis.dbsize()
                command_stats_dict = await redis.info('commandstats')
                command_stats = [
                    {'name': key.split('_', 1)[1], 'value': int(value.get('calls', 0))}
                    for key, value in sorted(command_stats_dict.items(), key=lambda item: item[0])
                ]
                return {
                    'ok': True,
                    'dbSize': db_size,
                    'commandStats': command_stats,
                    'cacheNames': self.domain_support.build_cache_name_items(),
                    'info': info,
                }
        except redis_error as exc:
            return self.redis_support.build_redis_error_result('读取缓存统计失败', exc)

    async def list_cache_keys(self, cache_name: str) -> dict[str, Any]:
        """
        列出指定缓存名称下的键名。

        :param cache_name: 缓存名称
        :return: 缓存键名列表结果
        """
        redis_error = self.infrastructure_gateway.get_redis_error_class()
        try:
            async with self.redis_support.redis_session() as (redis, _redis_util):
                cache_keys: list[str] = await redis.keys(self.redis_support.build_cache_name_keys_pattern(cache_name))
                key_items = self.domain_support.extract_cache_key_items(cache_name, cache_keys)
                return {'ok': True, 'cacheName': cache_name, 'count': len(key_items), 'keys': key_items}
        except redis_error as exc:
            return self.redis_support.build_redis_error_result('读取缓存键名失败', exc)

    async def get_cache_value(self, cache_name: str, cache_key: str) -> dict[str, Any]:
        """
        读取指定缓存键值。

        :param cache_name: 缓存名称
        :param cache_key: 缓存键名
        :return: 缓存值结果
        """
        redis_error = self.infrastructure_gateway.get_redis_error_class()
        full_cache_key = self.domain_support.build_full_cache_key(cache_name, cache_key)
        try:
            async with self.redis_support.redis_session() as (redis, _redis_util):
                cache_value = await redis.get(full_cache_key)
                if cache_value is None:
                    return self.domain_support.build_missing_cache_result(cache_name, cache_key)
                return {
                    'ok': True,
                    'cacheName': cache_name,
                    'cacheKey': cache_key,
                    'fullCacheKey': full_cache_key,
                    'cacheValue': cache_value,
                }
        except redis_error as exc:
            return self.redis_support.build_redis_error_result('读取缓存内容失败', exc)

    async def get_cache_ttl(self, cache_name: str, cache_key: str) -> dict[str, Any]:
        """
        读取指定缓存键的剩余过期时间。

        :param cache_name: 缓存名称
        :param cache_key: 缓存键名
        :return: 缓存 TTL 结果
        """
        redis_error = self.infrastructure_gateway.get_redis_error_class()
        full_cache_key = self.domain_support.build_full_cache_key(cache_name, cache_key)
        try:
            async with self.redis_support.redis_session() as (redis, _redis_util):
                ttl_seconds = await redis.ttl(full_cache_key)
                if ttl_seconds == REDIS_TTL_KEY_NOT_FOUND:
                    return self.domain_support.build_missing_cache_result(cache_name, cache_key)
                return {
                    'ok': True,
                    'message': '缓存剩余过期时间读取成功' if ttl_seconds >= 0 else '缓存存在且未设置过期时间',
                    'cacheName': cache_name,
                    'cacheKey': cache_key,
                    'fullCacheKey': full_cache_key,
                    'ttlSeconds': ttl_seconds,
                    'persistent': ttl_seconds == REDIS_TTL_PERSISTENT,
                    'expires': ttl_seconds >= 0,
                }
        except redis_error as exc:
            return self.redis_support.build_redis_error_result('读取缓存剩余过期时间失败', exc)

    async def clear_cache(
        self,
        *,
        cache_name: str = '',
        cache_key: str = '',
        clear_all: bool = False,
        dry_run: bool = False,
    ) -> dict[str, Any]:
        """
        清理缓存。

        :param cache_name: 需要按缓存名称清理的前缀
        :param cache_key: 需要按缓存键名模糊清理的键
        :param clear_all: 是否清理全部缓存
        :param dry_run: 是否仅演练执行
        :return: 缓存清理结果
        """
        selected_modes = [bool(cache_name), bool(cache_key), clear_all]
        if sum(selected_modes) != 1:
            return {
                'ok': False,
                'message': '必须且只能指定一种清理方式：--cache-name、--cache-key 或 --all',
                'exit_code': RUNTIME_ERROR,
            }

        redis_error = self.infrastructure_gateway.get_redis_error_class()
        try:
            async with self.redis_support.redis_session() as (redis, redis_util):
                target_pattern = self.redis_support.build_clear_target_pattern(
                    cache_name=cache_name,
                    cache_key=cache_key,
                    clear_all=clear_all,
                )
                if target_pattern is None:
                    target_keys = sorted(await redis.keys())
                else:
                    target_keys = sorted(await redis.keys(target_pattern))

                result = {
                    'ok': True,
                    'dryRun': dry_run,
                    'matchedCount': len(target_keys),
                    'matchedKeys': target_keys,
                    'scope': self.domain_support.build_clear_scope(
                        cache_name=cache_name,
                        cache_key=cache_key,
                        clear_all=clear_all,
                    ),
                }
                if dry_run:
                    result['message'] = '缓存清理演练完成，未执行实际删除'
                    return result

                if target_keys:
                    await redis.delete(*target_keys)

                if clear_all:
                    await redis_util.init_sys_dict(redis)
                    await redis_util.init_sys_config(redis)

                result['message'] = '缓存清理完成'
                return result
        except redis_error as exc:
            return self.redis_support.build_redis_error_result('清理缓存失败', exc)

    async def warmup_cache(self) -> dict[str, Any]:
        """
        预热系统缓存。

        :return: 缓存预热执行结果
        """
        redis_error = self.infrastructure_gateway.get_redis_error_class()
        try:
            async with self.redis_support.redis_session() as (redis, redis_util):
                await redis_util.init_sys_dict(redis)
                await redis_util.init_sys_config(redis)
                return {'ok': True, 'message': '缓存预热完成'}
        except redis_error as exc:
            return self.redis_support.build_redis_error_result('缓存预热失败', exc)


CACHE_RUNTIME = CacheRuntimeService()