63 lines
2.1 KiB
Python
63 lines
2.1 KiB
Python
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"]
|
|
|