LQConstPlanWriterAgent/foundation/infrastructure/cache/redis_config.py

# ！/usr/bin/python
# -*- coding: utf-8 -*-
'''
@Project    : lq-agent-api
@File       :redis_config.py
@IDE        :PyCharm
@Author     :
@Date       :2025/7/21 13:44
'''

from dataclasses import dataclass
from urllib.parse import quote
from foundation.infrastructure.config.config import config_handler


@dataclass
class RedisConfig:
    """Redis 连接配置"""
    url: str = "redis://127.0.0.1:6379"
    host: str = "127.0.0.1"
    port: int = 6379
    password: str = None
    db: int = 0
    max_connections: int = 50
    session_prefix: str = "session:"
    lock_prefix: str = "lock:"
    session_ttl: int = 3600  # 会话过期时间(秒)



def _build_redis_url(host: str, port: int, password: str | None, db: int) -> str:
    if password:
        return f"redis://:{quote(password, safe='')}@{host}:{port}/{db}"
    return f"redis://{host}:{port}/{db}"


def load_config_from_env() -> RedisConfig:
    """加载 Redis 配置，兼容 REDIS_URL 和 REDIS_HOST/PORT/PASSWORD/DB。"""
    host = config_handler.get("redis", "REDIS_HOST", "127.0.0.1")
    port = int(config_handler.get("redis", "REDIS_PORT", "6379"))
    password = config_handler.get("redis", "REDIS_PASSWORD", "") or None
    db = int(config_handler.get("redis", "REDIS_DB", "0"))
    redis_url = config_handler.get("redis", "REDIS_URL", "") or _build_redis_url(host, port, password, db)
    redis_config = RedisConfig(
        url=redis_url,
        host=host,
        port=port,
        password=password,
        db=db,
        max_connections=int(config_handler.get("redis", "REDIS_MAX_CONNECTIONS", "50"))
    )
    return redis_config