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