"""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()}")