""" Redis key-value operations with structured data handling. This module provides a class for managing Redis key-value operations with support for: - Structured data storage and retrieval - Key pattern generation for searches - JSON serialization/deserialization - Type-safe value handling """ import json from typing import Union, Dict, List, Optional, Any, ClassVar from datetime import datetime class RedisKeyError(Exception): """Exception raised for Redis key-related errors.""" pass class RedisValueError(Exception): """Exception raised for Redis value-related errors.""" pass class RedisRow: """ Handles Redis key-value operations with structured data. This class provides methods for: - Managing compound keys with delimiters - Converting between bytes and string formats - JSON serialization/deserialization of values - Pattern generation for Redis key searches Attributes: key: The Redis key in bytes or string format value: The stored value (will be JSON serialized) delimiter: Character used to separate compound key parts expires_at: Optional expiration timestamp """ key: ClassVar[Union[str, bytes]] value: ClassVar[Any] delimiter: ClassVar[str] = ":" expires_at: ClassVar[Optional[str]] = None @classmethod def merge(cls, set_values: List[Union[str, bytes]]) -> None: """ Merge list of values into a single delimited key. Args: set_values: List of values to merge into key Example: >>> RedisRow.merge(["users", "123", "profile"]) >>> print(RedisRow.key) b'users:123:profile' """ if not set_values: raise RedisKeyError("Cannot merge empty list of values") merged = [] for value in set_values: if value is None: continue if isinstance(value, bytes): value = value.decode() merged.append(str(value)) cls.key = cls.delimiter.join(merged).encode() @classmethod def regex(cls, list_keys: List[Union[str, bytes, None]]) -> str: """ Generate Redis search pattern from list of keys. Args: list_keys: List of key parts, can include None for wildcards Returns: str: Redis key pattern with wildcards Example: >>> RedisRow.regex([None, "users", "active"]) '*:users:active' """ if not list_keys: return "" # Filter and convert valid keys valid_keys = [] for key in list_keys: if key is None: continue if isinstance(key, bytes): key = key.decode() valid_keys.append(str(key)) # Build pattern pattern = cls.delimiter.join(valid_keys) if not pattern: return "" # Add wildcard if first key was None if list_keys[0] is None: pattern = f"*{cls.delimiter}{pattern}" return pattern @classmethod def parse(cls) -> List[str]: """ Parse the key into its component parts. Returns: List[str]: Key parts split by delimiter Example: >>> RedisRow.key = b'users:123:profile' >>> RedisRow.parse() ['users', '123', 'profile'] """ if not cls.key: return [] key_str = cls.key.decode() if isinstance(cls.key, bytes) else cls.key return key_str.split(cls.delimiter) @classmethod def feed(cls, value: Union[bytes, Dict, List]) -> None: """ Convert and store value in JSON format. Args: value: Value to store (bytes, dict, or list) Raises: RedisValueError: If value type is not supported Example: >>> RedisRow.feed({"name": "John", "age": 30}) >>> print(RedisRow.value) '{"name": "John", "age": 30}' """ try: if isinstance(value, (dict, list)): cls.value = json.dumps(value) elif isinstance(value, bytes): cls.value = json.dumps(json.loads(value.decode())) else: raise RedisValueError(f"Unsupported value type: {type(value)}") except json.JSONDecodeError as e: raise RedisValueError(f"Invalid JSON format: {str(e)}") @classmethod def modify(cls, add_dict: Dict) -> None: """ Modify existing data by merging with new dictionary. Args: add_dict: Dictionary to merge with existing data Example: >>> RedisRow.feed({"name": "John"}) >>> RedisRow.modify({"age": 30}) >>> print(RedisRow.data) {"name": "John", "age": 30} """ if not isinstance(add_dict, dict): raise RedisValueError("modify() requires a dictionary argument") current_data = cls.data if cls.data else {} if not isinstance(current_data, dict): raise RedisValueError("Cannot modify non-dictionary data") cls.feed({**current_data, **add_dict}) @classmethod def remove(cls, key: str) -> None: """ Remove a key from the stored dictionary. Args: key: Key to remove from stored dictionary Raises: KeyError: If key doesn't exist RedisValueError: If stored value is not a dictionary """ current_data = cls.data if not isinstance(current_data, dict): raise RedisValueError("Cannot remove key from non-dictionary data") try: current_data.pop(key) cls.feed(current_data) except KeyError: raise KeyError(f"Key '{key}' not found in stored data") @property def keys(self) -> str: """ Get key as string. Returns: str: Key in string format """ return self.key.decode() if isinstance(self.key, bytes) else self.key @classmethod def set_key(cls, key: Union[str, bytes]) -> None: """ Set key ensuring bytes format. Args: key: Key in string or bytes format """ if not key: raise RedisKeyError("Cannot set empty key") cls.key = key if isinstance(key, bytes) else str(key).encode() @property def redis_key(self) -> bytes: """ Get key in bytes format for Redis operations. Returns: bytes: Key in bytes format """ return self.key if isinstance(self.key, bytes) else str(self.key).encode() @property def data(self) -> Union[Dict, List]: """ Get stored value as Python object. Returns: Union[Dict, List]: Deserialized JSON data """ try: return json.loads(self.value) except json.JSONDecodeError as e: raise RedisValueError(f"Invalid JSON format in stored value: {str(e)}") @property def as_dict(self) -> Dict[str, Any]: """ Get row data as dictionary. Returns: Dict[str, Any]: Dictionary with keys and value """ return { "keys": self.keys, "value": self.data, }