Blame for schema_registry.py

"""Schema Registry - Mapping/Audit Utility

Provides schema registration, validation, and mapping capabilities
with audit trail and versioning support.
"""

import json
import csv
from typing import Dict, List, Any, Optional, Tuple
from datetime import datetime
import logging
from pathlib import Path

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)


class SchemaRegistry:
    """Manages schema definitions, mappings, and validation."""
    
    def __init__(self, registry_path: str = "schema_registry.json", log_path: str = "schema_audit.log"):
        self.registry_path = Path(registry_path)
        self.log_path = Path(log_path)
        self.schemas: Dict[str, Dict[str, Any]] = {}
        self.mappings: Dict[str, Dict[str, str]] = {}
        self.version = "1.0.0"
        self._load_registry()
        
    def _load_registry(self) -> None:
        """Load existing schema registry from file."""
        if self.registry_path.exists():
            try:
                with open(self.registry_path, 'r') as f:
                    data = json.load(f)
                    self.schemas = data.get('schemas', {})
                    self.mappings = data.get('mappings', {})
                    self.version = data.get('version', '1.0.0')
                logger.info(f"Loaded registry with {len(self.schemas)} schemas")
            except Exception as e:
                logger.error(f"Failed to load registry: {e}")
                self._audit_log("ERROR", f"Registry load failed: {e}")
        else:
            logger.info("No existing registry found, starting fresh")
            
    def _save_registry(self) -> None:
        """Save schema registry to file."""
        try:
            data = {
                'version': self.version,
                'schemas': self.schemas,
                'mappings': self.mappings,
                'last_updated': datetime.now().isoformat()
            }
            with open(self.registry_path, 'w') as f:
                json.dump(data, f, indent=2)
            logger.info("Registry saved successfully")
        except Exception as e:
            logger.error(f"Failed to save registry: {e}")
            self._audit_log("ERROR", f"Registry save failed: {e}")
            
    def _audit_log(self, level: str, message: str, metadata: Optional[Dict] = None) -> None:
        """Write audit log entry."""
        timestamp = datetime.now().isoformat()
        log_entry = {
            'timestamp': timestamp,
            'level': level,
            'message': message,
            'metadata': metadata or {}
        }
        
        try:
            with open(self.log_path, 'a') as f:
                f.write(json.dumps(log_entry) + '\n')
        except Exception as e:
            logger.error(f"Failed to write audit log: {e}")
            
    def register_schema(self, name: str, schema: Dict[str, Any], description: str = "") -> bool:
        """Register a new schema definition.
        
        Args:
            name: Unique schema identifier
            schema: Schema definition with field types and constraints
            description: Human-readable schema description
            
        Returns:
            True if successful, False otherwise
        """
        try:
            if name in self.schemas:
                logger.warning(f"Schema '{name}' already exists, will be overwritten")
                
            self.schemas[name] = {
                'schema': schema,
                'description': description,
                'created_at': datetime.now().isoformat(),
                'version': self.version
            }
            
            self._save_registry()
            self._audit_log("INFO", f"Registered schema: {name}", {'schema_name': name})
            return True
            
        except Exception as e:
            logger.error(f"Failed to register schema '{name}': {e}")
            self._audit_log("ERROR", f"Schema registration failed: {name}", {'error': str(e)})
            return False
            
    def register_mapping(self, source_schema: str, target_schema: str, 
                        field_mappings: Dict[str, str]) -> bool:
        """Register field mappings between two schemas.
        
        Args:
            source_schema: Source schema name
            target_schema: Target schema name
            field_mappings: Dict mapping source fields to target fields
            
        Returns:
            True if successful, False otherwise
        """
        try:
            mapping_key = f"{source_schema}->{target_schema}"
            
            # Validate schemas exist
            if source_schema not in self.schemas:
                raise ValueError(f"Source schema '{source_schema}' not found")
            if target_schema not in self.schemas:
                raise ValueError(f"Target schema '{target_schema}' not found")
                
            # Validate field mappings
            source_fields = self.schemas[source_schema]['schema'].keys()
            target_fields = self.schemas[target_schema]['schema'].keys()
            
            for src_field, tgt_field in field_mappings.items():
                if src_field not in source_fields:
                    raise ValueError(f"Source field '{src_field}' not in schema '{source_schema}'")
                if tgt_field not in target_fields:
                    raise ValueError(f"Target field '{tgt_field}' not in schema '{target_schema}'")
                    
            self.mappings[mapping_key] = {
                'field_mappings': field_mappings,
                'created_at': datetime.now().isoformat()
            }
            
            self._save_registry()
            self._audit_log("INFO", f"Registered mapping: {mapping_key}", 
                          {'source': source_schema, 'target': target_schema})
            return True
            
        except Exception as e:
            logger.error(f"Failed to register mapping: {e}")
            self._audit_log("ERROR", f"Mapping registration failed", {'error': str(e)})
            return False
            
    def validate_data(self, schema_name: str, data: Dict[str, Any]) -> Tuple[bool, List[str]]:
        """Validate data against a registered schema.
        
        Args:
            schema_name: Name of schema to validate against
            data: Data to validate
            
        Returns:
            Tuple of (is_valid, list of error messages)
        """
        errors = []
        
        try:
            if schema_name not in self.schemas:
                return False, [f"Schema '{schema_name}' not found"]
                
            schema = self.schemas[schema_name]['schema']
            
            # Check required fields
            for field, field_def in schema.items():
                if field_def.get('required', False) and field not in data:
                    errors.append(f"Required field '{field}' is missing")
                    
                # Type validation if field exists
                if field in data:
                    expected_type = field_def.get('type')
                    actual_value = data[field]
                    
                    if expected_type and not self._check_type(actual_value, expected_type):
                        errors.append(f"Field '{field}' has wrong type. Expected {expected_type}, got {type(actual_value).__name__}")
                        
                    # Additional constraints
                    if 'min_length' in field_def and isinstance(actual_value, str):
                        if len(actual_value) < field_def['min_length']:
                            errors.append(f"Field '{field}' is too short (min: {field_def['min_length']})")
                            
                    if 'max_length' in field_def and isinstance(actual_value, str):
                        if len(actual_value) > field_def['max_length']:
                            errors.append(f"Field '{field}' is too long (max: {field_def['max_length']})")
                            
            is_valid = len(errors) == 0
            self._audit_log("INFO", f"Validation {'passed' if is_valid else 'failed'} for schema: {schema_name}", 
                          {'schema': schema_name, 'valid': is_valid, 'error_count': len(errors)})
            
            return is_valid, errors
            
        except Exception as e:
            error_msg = f"Validation error: {str(e)}"
            logger.error(error_msg)
            return False, [error_msg]
            
    def _check_type(self, value: Any, expected_type: str) -> bool:
        """Check if value matches expected type."""
        type_map = {
            'string': str,
            'int': int,
            'float': float,
            'bool': bool,
            'list': list,
            'dict': dict,
            'any': object
        }
        
        expected_class = type_map.get(expected_type.lower())
        if expected_class:
            return isinstance(value, expected_class)
        return True
        
    def transform_data(self, data: Dict[str, Any], source_schema: str, 
                      target_schema: str) -> Tuple[Optional[Dict[str, Any]], List[str]]:
        """Transform data from source schema to target schema using registered mappings.
        
        Args:
            data: Source data
            source_schema: Source schema name
            target_schema: Target schema name
            
        Returns:
            Tuple of (transformed_data or None, list of errors)
        """
        errors = []
        
        try:
            # Validate source data
            is_valid, validation_errors = self.validate_data(source_schema, data)
            if not is_valid:
                return None, validation_errors
                
            # Get mapping
            mapping_key = f"{source_schema}->{target_schema}"
            if mapping_key not in self.mappings:
                return None, [f"No mapping found for {mapping_key}"]
                
            field_mappings = self.mappings[mapping_key]['field_mappings']
            
            # Transform data
            transformed = {}
            for src_field, tgt_field in field_mappings.items():
                if src_field in data:
                    transformed[tgt_field] = data[src_field]
                    
            # Validate transformed data
            is_valid, validation_errors = self.validate_data(target_schema, transformed)
            if not is_valid:
                return None, validation_errors
                
            self._audit_log("INFO", f"Data transformation successful: {mapping_key}", 
                          {'source': source_schema, 'target': target_schema})
            
            return transformed, []
            
        except Exception as e:
            error_msg = f"Transformation error: {str(e)}"
            logger.error(error_msg)
            self._audit_log("ERROR", "Data transformation failed", {'error': str(e)})
            return None, [error_msg]
            
    def get_schema(self, name: str) -> Optional[Dict[str, Any]]:
        """Retrieve a registered schema."""
        return self.schemas.get(name)
        
    def list_schemas(self) -> List[str]:
        """List all registered schema names."""
        return list(self.schemas.keys())
        
    def list_mappings(self) -> List[str]:
        """List all registered mappings."""
        return list(self.mappings.keys())
        
    def export_to_json(self, filepath: str) -> bool:
        """Export registry to JSON file."""
        try:
            data = {
                'version': self.version,
                'schemas': self.schemas,
                'mappings': self.mappings,
                'exported_at': datetime.now().isoformat()
            }
            with open(filepath, 'w') as f:
                json.dump(data, f, indent=2)
            logger.info(f"Registry exported to {filepath}")
            return True
        except Exception as e:
            logger.error(f"Export failed: {e}")
            return False
            
    def import_from_json(self, filepath: str) -> bool:
        """Import registry from JSON file."""
        try:
            with open(filepath, 'r') as f:
                data = json.load(f)
                self.schemas.update(data.get('schemas', {}))
                self.mappings.update(data.get('mappings', {}))
            self._save_registry()
            logger.info(f"Registry imported from {filepath}")
            return True
        except Exception as e:
            logger.error(f"Import failed: {e}")
            return False


if __name__ == "__main__":
    # Example usage
    registry = SchemaRegistry()
    
    # Register example schemas
    user_schema = {
        'id': {'type': 'int', 'required': True},
        'username': {'type': 'string', 'required': True, 'min_length': 3},
        'email': {'type': 'string', 'required': True},
        'age': {'type': 'int', 'required': False}
    }
    
    registry.register_schema('user', user_schema, 'User profile schema')
    
    print(f"Registered schemas: {registry.list_schemas()}")