string_to_dynamodb_expression.py

#!/usr/bin/env python3
"""
Production-ready String to DynamoDB Expression Converter

Safely converts string representations of DynamoDB Attr expressions 
to actual boto3 Attr expression objects without using eval().

Example:
    "Attr('name').eq('san') & Attr('age').eq(40)" 
    -> Real boto3 Attr expression object
"""

import re
import logging
from typing import Any, List, Union, Optional

try:
    from boto3.dynamodb.conditions import Attr
except ImportError as e:
    raise ImportError(
        "boto3 is required for DynamoDB operations. "
        "Install with: pip install boto3"
    ) from e

# Set up logging
logger = logging.getLogger(__name__)


class StringToDynamoExpression:
    """
    Convert string representations of DynamoDB Attr expressions 
    to actual boto3 Attr expression objects safely.
    
    This class provides a secure way to parse and convert string-based
    DynamoDB filter expressions without using eval() or other unsafe methods.
    """
    
    def __init__(self):
        """Initialize the converter with supported operators."""
        # Supported comparison operators
        self.operators = {
            'eq': 'eq',           # Equal
            'ne': 'ne',           # Not equal
            'lt': 'lt',           # Less than
            'lte': 'lte',         # Less than or equal
            'gt': 'gt',           # Greater than
            'gte': 'gte',         # Greater than or equal
            'between': 'between', # Between two values
            'begins_with': 'begins_with',  # Begins with
            'contains': 'contains',        # Contains
            'is_in': 'is_in',             # In list
            'exists': 'exists',           # Attribute exists
            'not_exists': 'not_exists',   # Attribute doesn't exist
            'size': 'size'               # Size function
        }
        
        # Logical operators patterns
        self.logical_patterns = {
            '&': 'and',
            '|': 'or'
        }
    
    def convert(self, expression_string: str) -> Any:
        """
        Convert string to DynamoDB Attr expression.
        
        Args:
            expression_string: String like "Attr('name').eq('san') & Attr('age').eq(40)"
            
        Returns:
            boto3 Attr expression object
            
        Raises:
            ValueError: If string format is invalid or empty
            TypeError: If input is not a string
        """
        # Input validation
        if not isinstance(expression_string, str):
            raise TypeError(f"Expression must be a string, got {type(expression_string)}")
        
        if not expression_string or not expression_string.strip():
            raise ValueError("Expression string cannot be empty")
        
        # Log the conversion attempt
        logger.debug(f"Converting expression: {expression_string}")
        
        # Clean the input string
        clean_string = expression_string.strip()
        
        # Handle single condition (no logical operators)
        if '&' not in clean_string and '|' not in clean_string:
            return self._parse_single_condition(clean_string)
        
        # Handle multiple conditions with logical operators
        return self._parse_multiple_conditions(clean_string)
    
    def _parse_multiple_conditions(self, expression_string: str) -> Any:
        """Parse expression with multiple conditions and logical operators."""
        try:
            # Split by logical operators while preserving operator info
            parts = []
            logical_ops = []
            current_part = ""
            
            i = 0
            while i < len(expression_string):
                char = expression_string[i]
                
                if char in ['&', '|']:
                    # Found logical operator
                    parts.append(current_part.strip())
                    logical_ops.append(self.logical_patterns[char])
                    current_part = ""
                else:
                    current_part += char
                i += 1
            
            # Add the last part
            if current_part.strip():
                parts.append(current_part.strip())
            
            # Parse each condition
            conditions = []
            for part in parts:
                if part:  # Skip empty parts
                    condition = self._parse_single_condition(part)
                    conditions.append(condition)
            
            # Validate we have conditions
            if not conditions:
                raise ValueError("No valid conditions found in expression")
            
            # Combine conditions with logical operators
            if len(conditions) == 1:
                return conditions[0]
            
            result = conditions[0]
            for i, condition in enumerate(conditions[1:]):
                if i < len(logical_ops):
                    if logical_ops[i] == 'and':
                        result = result & condition
                    elif logical_ops[i] == 'or':
                        result = result | condition
                else:
                    # Default to 'and' if no logical operator specified
                    result = result & condition
            
            return result
            
        except Exception as e:
            logger.error(f"Error parsing multiple conditions: {e}")
            raise ValueError(f"Failed to parse expression: {e}") from e
    
    def _parse_single_condition(self, condition_string: str) -> Any:
        """Parse a single Attr condition like Attr('name').eq('san')"""
        try:
            condition_string = condition_string.strip()
            
            # Validate input
            if not condition_string:
                raise ValueError("Condition string cannot be empty")
            
            # Pattern to match Attr('param').operator(value)
            # Handle both single and double quotes
            attr_pattern = r"Attr\(['\"]([^'\"]+)['\"]\)\.([a-zA-Z_][a-zA-Z0-9_]*)\(([^)]*)\)"
            
            match = re.search(attr_pattern, condition_string)
            if not match:
                raise ValueError(f"Invalid Attr expression format: {condition_string}")
            
            param_name = match.group(1).strip()
            operator = match.group(2).strip()
            value_str = match.group(3)
            
            # Validate parameter name
            if not param_name:
                raise ValueError("Parameter name cannot be empty")
            
            # Validate operator
            if operator not in self.operators:
                raise ValueError(
                    f"Unsupported operator '{operator}'. "
                    f"Supported: {list(self.operators.keys())}"
                )
            
            # Parse value(s)
            values = self._parse_values(value_str)
            
            # Build the Attr expression
            attr_obj = Attr(param_name)
            attr_method = getattr(attr_obj, operator)
            
            # Handle different value patterns
            if len(values) == 1:
                return attr_method(values[0])
            else:
                # Multiple values (for operators like between, is_in)
                return attr_method(*values)
                
        except Exception as e:
            logger.error(f"Error parsing condition '{condition_string}': {e}")
            raise ValueError(f"Failed to parse condition: {e}") from e
    
    def _parse_values(self, value_string: str) -> List[Any]:
        """Parse value(s) from string safely (no eval!)"""
        try:
            value_string = value_string.strip()
            
            # Handle empty values
            if not value_string:
                return [None]
            
            # Handle multiple values (comma-separated)
            if ',' in value_string:
                # Split by comma and parse each value
                raw_values = [v.strip() for v in value_string.split(',')]
                return [self._parse_single_value(v) for v in raw_values if v.strip()]
            else:
                # Single value
                return [self._parse_single_value(value_string)]
                
        except Exception as e:
            logger.error(f"Error parsing values '{value_string}': {e}")
            raise ValueError(f"Failed to parse values: {e}") from e
    
    def _parse_single_value(self, value_str: str) -> Any:
        """Parse a single value string to appropriate Python type."""
        if not isinstance(value_str, str):
            raise TypeError(f"Value must be a string, got {type(value_str)}")
            
        value_str = value_str.strip()
        
        # Handle empty strings
        if not value_str:
            return None
        
        # Handle quoted strings (single or double quotes)
        if (value_str.startswith("'") and value_str.endswith("'")) or \
           (value_str.startswith('"') and value_str.endswith('"')):
            if len(value_str) < 2:
                raise ValueError(f"Invalid quoted string: {value_str}")
            return value_str[1:-1]  # Remove quotes
        
        # Handle boolean values
        if value_str.lower() == 'true':
            return True
        if value_str.lower() == 'false':
            return False
        
        # Handle None/null values
        if value_str.lower() in ('none', 'null'):
            return None
        
        # Handle numeric values
        try:
            # Try integer first
            if '.' not in value_str and 'e' not in value_str.lower():
                return int(value_str)
            else:
                return float(value_str)
        except ValueError:
            pass
        
        # Handle lists (for is_in operator)
        if value_str.startswith('[') and value_str.endswith(']'):
            try:
                # Simple list parsing (no nested structures)
                list_content = value_str[1:-1].strip()
                if not list_content:
                    return []
                items = [item.strip() for item in list_content.split(',')]
                return [self._parse_single_value(item) for item in items if item.strip()]
            except Exception as e:
                raise ValueError(f"Invalid list format: {value_str}") from e
        
        # If all else fails, treat as unquoted string
        return value_str


# Production-ready usage example
if __name__ == "__main__":
    # Configure logging for production
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    
    try:
        converter = StringToDynamoExpression()
        
        # Test the converter
        test_expression = "Attr('name').eq('san') & Attr('age').eq(40)"
        result = converter.convert(test_expression)
        
        logger.info(f"Successfully converted: {test_expression}")
        logger.info(f"Result type: {type(result)}")
        
    except Exception as e:
        logger.error(f"Error in main: {e}")
        raise