import json import logging from typing import TYPE_CHECKING, Any, Optional, Set, Union import redis.asyncio as aioredis from redis.asyncio import Redis if TYPE_CHECKING: pass # type: ignore[attr-defined] from settings import REDIS_URL from utils.logger import root_logger as logger logger = logging.getLogger(__name__) # Set redis logging level to suppress DEBUG messages redis_logger = logging.getLogger("redis") redis_logger.setLevel(logging.WARNING) class RedisService: """ Сервис для работы с Redis с поддержкой пулов соединений. Provides connection pooling and proper error handling for Redis operations. """ def __init__(self, redis_url: str = REDIS_URL) -> None: self._client: Optional[Redis[Any]] = None self._redis_url = redis_url self._is_available = aioredis is not None if not self._is_available: logger.warning("Redis is not available - aioredis not installed") async def connect(self) -> None: """Establish Redis connection""" if not self._is_available: return # Закрываем существующее соединение если есть if self._client: try: await self._client.close() except Exception: pass self._client = None try: self._client = aioredis.from_url( self._redis_url, encoding="utf-8", decode_responses=False, # We handle decoding manually socket_keepalive=True, socket_keepalive_options={}, retry_on_timeout=True, health_check_interval=30, socket_connect_timeout=5, socket_timeout=5, ) # Test connection await self._client.ping() logger.info("Successfully connected to Redis") except Exception as e: logger.error(f"Failed to connect to Redis: {e}") if self._client: try: await self._client.close() except Exception: pass self._client = None async def disconnect(self) -> None: """Close Redis connection""" if self._client: await self._client.close() self._client = None @property def is_connected(self) -> bool: """Check if Redis is connected""" return self._client is not None and self._is_available def pipeline(self) -> Any: # Returns Pipeline but we can't import it safely """Create a Redis pipeline""" if self._client: return self._client.pipeline() return None async def execute(self, command: str, *args: Any) -> Any: """Execute a Redis command""" if not self._is_available: logger.debug(f"Redis not available, skipping command: {command}") return None # Проверяем и восстанавливаем соединение при необходимости if not self.is_connected: logger.info("Redis not connected, attempting to reconnect...") await self.connect() if not self.is_connected: logger.error(f"Failed to establish Redis connection for command: {command}") return None try: # Get the command method from the client cmd_method = getattr(self._client, command.lower(), None) if cmd_method is None: logger.error(f"Unknown Redis command: {command}") return None result = await cmd_method(*args) return result except (ConnectionError, AttributeError, OSError) as e: logger.warning(f"Redis connection lost during {command}, attempting to reconnect: {e}") # Попытка переподключения await self.connect() if self.is_connected: try: cmd_method = getattr(self._client, command.lower(), None) if cmd_method is not None: result = await cmd_method(*args) return result except Exception as retry_e: logger.error(f"Redis retry failed for {command}: {retry_e}") return None except Exception as e: logger.error(f"Redis command failed {command}: {e}") return None async def get(self, key: str) -> Optional[Union[str, bytes]]: """Get value by key""" return await self.execute("get", key) async def set(self, key: str, value: Any, ex: Optional[int] = None) -> bool: """Set key-value pair with optional expiration""" if ex is not None: result = await self.execute("setex", key, ex, value) else: result = await self.execute("set", key, value) return result is not None async def delete(self, *keys: str) -> int: """Delete keys""" result = await self.execute("delete", *keys) return result or 0 async def exists(self, key: str) -> bool: """Check if key exists""" result = await self.execute("exists", key) return bool(result) async def publish(self, channel: str, data: Any) -> None: """Publish message to channel""" if not self.is_connected or self._client is None: logger.debug(f"Redis not available, skipping publish to {channel}") return try: await self._client.publish(channel, data) except Exception as e: logger.error(f"Failed to publish to channel {channel}: {e}") async def hset(self, key: str, field: str, value: Any) -> None: """Set hash field""" await self.execute("hset", key, field, value) async def hget(self, key: str, field: str) -> Optional[Union[str, bytes]]: """Get hash field""" return await self.execute("hget", key, field) async def hgetall(self, key: str) -> dict[str, Any]: """Get all hash fields""" result = await self.execute("hgetall", key) return result or {} async def keys(self, pattern: str) -> list[str]: """Get keys matching pattern""" result = await self.execute("keys", pattern) return result or [] async def smembers(self, key: str) -> Set[str]: """Get set members""" if not self.is_connected or self._client is None: return set() try: result = await self._client.smembers(key) if result: return {str(item.decode("utf-8") if isinstance(item, bytes) else item) for item in result} return set() except Exception as e: logger.error(f"Redis smembers command failed for {key}: {e}") return set() async def sadd(self, key: str, *members: str) -> int: """Add members to set""" result = await self.execute("sadd", key, *members) return result or 0 async def srem(self, key: str, *members: str) -> int: """Remove members from set""" result = await self.execute("srem", key, *members) return result or 0 async def expire(self, key: str, seconds: int) -> bool: """Set key expiration""" result = await self.execute("expire", key, seconds) return bool(result) async def serialize_and_set(self, key: str, data: Any, ex: Optional[int] = None) -> bool: """Serialize data to JSON and store in Redis""" try: if isinstance(data, (str, bytes)): serialized_data: bytes = data.encode("utf-8") if isinstance(data, str) else data else: serialized_data = json.dumps(data).encode("utf-8") return await self.set(key, serialized_data, ex=ex) except Exception as e: logger.error(f"Failed to serialize and set {key}: {e}") return False async def get_and_deserialize(self, key: str) -> Any: """Get data from Redis and deserialize from JSON""" try: data = await self.get(key) if data is None: return None if isinstance(data, bytes): data = data.decode("utf-8") return json.loads(data) except Exception as e: logger.error(f"Failed to get and deserialize {key}: {e}") return None async def ping(self) -> bool: """Ping Redis server""" if not self.is_connected or self._client is None: return False try: result = await self._client.ping() return bool(result) except Exception: return False async def execute_pipeline(self, commands: list[tuple[str, tuple[Any, ...]]]) -> list[Any]: """ Выполняет список команд через pipeline для лучшей производительности. Избегает использования async context manager для pipeline чтобы избежать deprecated warnings. Args: commands: Список кортежей (команда, аргументы) Returns: Список результатов выполнения команд """ if not self.is_connected or self._client is None: logger.warning("Redis not connected, cannot execute pipeline") return [] try: pipe = self.pipeline() if pipe is None: logger.error("Failed to create Redis pipeline") return [] # Добавляем команды в pipeline for command, args in commands: cmd_method = getattr(pipe, command.lower(), None) if cmd_method is not None: cmd_method(*args) else: logger.error(f"Unknown Redis command in pipeline: {command}") # Выполняем pipeline results = await pipe.execute() return results except Exception as e: logger.error(f"Redis pipeline execution failed: {e}") return [] # Global Redis instance redis = RedisService() async def init_redis() -> None: """Initialize Redis connection""" await redis.connect() async def close_redis() -> None: """Close Redis connection""" await redis.disconnect()