from pydantic.generics import GenericModel import os import redis class RedisSettings(GenericModel): REDIS_HOST: str = os.getenv("REDIS_HOST", "192.168.1.19") REDIS_PORT: int = int(os.getenv("REDIS_PORT", 6379)) REDIS_DB: int = int(os.getenv("REDIS_DB", 0)) REDIS_PASSWORD: str = os.getenv("REDIS_PASSWORD") or None @property def REDIS_URL(self) -> str: """Получить Redis URL для подключения""" if self.REDIS_PASSWORD: return f"redis://:{self.REDIS_PASSWORD}@{self.REDIS_HOST}:{self.REDIS_PORT}/{self.REDIS_DB}" else: return f"redis://{self.REDIS_HOST}:{self.REDIS_PORT}/{self.REDIS_DB}" @property def connection_kwargs(self) -> dict: """Получить параметры подключения для redis.Redis()""" return { "host": self.REDIS_HOST, "port": self.REDIS_PORT, "db": self.REDIS_DB, "password": self.REDIS_PASSWORD, "decode_responses": True, } def get_client(self) -> redis.Redis: """Создать и вернуть Redis клиент""" return redis.Redis(**self.connection_kwargs) def test_connection(self) -> bool: """Протестировать подключение к Redis""" try: client = self.get_client() client.ping() print(f"✓ Успешно подключено к Redis на {self.REDIS_HOST}:{self.REDIS_PORT}") return True except redis.ConnectionError as e: print(f"✗ Ошибка подключения к Redis: {e}") return False # Глобальная конфигурация _redis_settings = RedisSettings() def get_redis_settings() -> RedisSettings: """Получить конфигурацию Redis""" return _redis_settings def get_redis_client() -> redis.Redis: """Получить готовый клиент Redis""" return _redis_settings.get_client() __all__ = ["RedisSettings", "get_redis_settings", "get_redis_client"]