""" 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 arrow import json from connection import redis_cli from typing import Union, Dict, List, Optional, Any, ClassVar 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: Union[str, bytes] value: Optional[str] = None delimiter: str = ":" expires_at: Optional[dict] = {"seconds": 60 * 60 * 30} expires_at_string: Optional[str] def get_expiry_time(self) -> int | None: """Calculate expiry time in seconds from kwargs.""" time_multipliers = {"days": 86400, "hours": 3600, "minutes": 60, "seconds": 1} if self.expires_at: return sum( int(self.expires_at.get(unit, 0)) * multiplier for unit, multiplier in time_multipliers.items() ) return None def merge(self, 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)) self.key = self.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 or str(key) == "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}" if "*" not in pattern and any([list_key is None for list_key in list_keys]): pattern = f"{pattern}:*" return pattern def parse(self) -> 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 self.key: return [] key_str = self.key.decode() if isinstance(self.key, bytes) else self.key return key_str.split(self.delimiter) def feed(self, value: Union[bytes, Dict, List, str]) -> 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)): self.value = json.dumps(value) elif isinstance(value, bytes): self.value = json.dumps(json.loads(value.decode())) elif isinstance(value, str): self.value = value else: raise RedisValueError(f"Unsupported value type: {type(value)}") except json.JSONDecodeError as e: raise RedisValueError(f"Invalid JSON format: {str(e)}") def modify(self, 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 = self.row if self.row else {} if not isinstance(current_data, dict): raise RedisValueError("Cannot modify non-dictionary data") current_data = { **current_data, **add_dict, } self.feed(current_data) self.save() def save(self): """ Save the data to Redis with optional expiration. Raises: RedisKeyError: If key is not set RedisValueError: If value is not set """ if not self.key: raise RedisKeyError("Cannot save data without a key") if not self.value: raise RedisValueError("Cannot save empty data") if self.expires_at: redis_cli.setex( name=self.redis_key, time=self.get_expiry_time(), value=self.value ) self.expires_at_string = str( arrow.now() .shift(seconds=self.get_expiry_time()) .format("YYYY-MM-DD HH:mm:ss") ) return self.value redis_cli.set(name=self.redis_key, value=self.value) self.expires_at = None self.expires_at_string = None return self.value def remove(self, 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 = self.row if not isinstance(current_data, dict): raise RedisValueError("Cannot remove key from non-dictionary data") try: current_data.pop(key) self.feed(current_data) self.save() except KeyError: raise KeyError(f"Key '{key}' not found in stored data") def delete(self) -> None: """Delete the key from Redis.""" try: redis_cli.delete(self.redis_key) except Exception as e: print(f"Error deleting key: {str(e)}") @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 def set_key(self, 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") self.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 row(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.row, }