Add schema registry with mapping, validation, and audit capabilities

schema_registry.py

@@ -0,0 +1,330 @@
+"""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()}")