hts-code-query-system-backend/config/get_redis.py

# _*_ coding : UTF-8 _*_
# @Time : 2025/01/18 02:41
# @UpdateTime : 2025/01/18 02:41
# @Author : sonder
# @File : get_redis.py
# @Software : PyCharm
# @Comment : 本程序

from redis import asyncio as aioredis
from redis.exceptions import AuthenticationError, TimeoutError, RedisError

from config.constant import RedisKeyConfig
from config.env import RedisConfig
from models import Config
from utils.log import logger


class Redis:
    """
    Redis工具类
    提供与 Redis 交互的常用方法，包括连接管理和缓存初始化。
    """

    @classmethod
    async def create_redis_pool(cls) -> aioredis.Redis:
        """
        应用启动时初始化 Redis 连接池。

        该方法通过 `aioredis` 创建一个 Redis 连接池，并验证连接是否成功。
        如果连接失败，会捕获异常并记录错误日志。

        :return: Redis 连接对象
        """
        logger.info('开始连接 Redis...')
        redis = await aioredis.from_url(
            url=f'redis://{RedisConfig.redis_host}',
            port=RedisConfig.redis_port,
            username=RedisConfig.redis_username,
            password=RedisConfig.redis_password,
            db=RedisConfig.redis_database,
            encoding='utf-8',  # 默认编码为 UTF-8
            decode_responses=True,  # 自动解码 Redis 响应
        )
        try:
            # 测试 Redis 连接
            connection = await redis.ping()
            if connection:
                logger.info('Redis 连接成功')
            else:
                logger.error('Redis 连接失败')
        except AuthenticationError as e:
            logger.error(f'Redis 用户名或密码错误，详细错误信息：{e}')
        except TimeoutError as e:
            logger.error(f'Redis 连接超时，详细错误信息：{e}')
        except RedisError as e:
            logger.error(f'Redis 连接错误，详细错误信息：{e}')
        return redis

    @classmethod
    async def close_redis_pool(cls, app):
        """
        应用关闭时关闭 Redis 连接池。

        在应用生命周期结束时，清理 Redis 连接池以释放资源。

        :param app: FastAPI 应用对象，用于访问全局 Redis 连接池。
        :return: None
        """
        await app.state.redis.close()
        logger.info('关闭 Redis 连接成功')

    @classmethod
    async def init_system_config(cls, app):
        """
        初始化系统配置
        """
        # 获取以sys_config:开头的键列表
        keys = await app.state.redis.keys(f'{RedisKeyConfig.SYSTEM_CONFIG.key}:*')
        # 删除匹配的键
        if keys:
            await app.state.redis.delete(*keys)
        config = await Config.filter(del_flag=1).values()
        for item in config:
            await app.state.redis.set(f"{RedisKeyConfig.SYSTEM_CONFIG.key}:{item.get('key')}",
                                      item.get('value'), )