# -*- coding: utf-8 -*- """ Schema-basierte Validierung für komplexe Datenstrukturen. Bietet deklarative Schema-Definitionen für die Validierung von Dictionaries und Dataclasses. """ from __future__ import annotations import dataclasses from dataclasses import dataclass, field as dataclass_field from typing import Any, Callable, Generic, Sequence, Type, TypeVar, get_type_hints from trixy_core.utils.validation.errors import ( FieldError, SchemaError, ValidationError, ValidationErrors, ) from trixy_core.utils.validation.validators import ( ListValidator, RequiredValidator, TypeValidator, Validator, ) T = TypeVar("T") @dataclass class Field: """ Definition eines Schema-Feldes. Beschreibt die Validierungsregeln für ein einzelnes Feld in einem Schema. """ name: str """Name des Feldes.""" validators: list[Validator] = dataclass_field(default_factory=list) """Liste von Validatoren für das Feld.""" required: bool = True """Ob das Feld erforderlich ist.""" default: Any = None """Standardwert falls nicht vorhanden.""" default_factory: Callable[[], Any] | None = None """Factory-Funktion für den Standardwert.""" description: str = "" """Beschreibung des Feldes.""" alias: str | None = None """Alternativer Name im Input.""" coerce: Callable[[Any], Any] | None = None """Funktion zur Typkonvertierung vor Validierung.""" def get_default(self) -> Any: """ Gibt den Standardwert zurück. Returns: Standardwert oder Ergebnis der Factory. """ if self.default_factory: return self.default_factory() return self.default def validate(self, value: Any) -> None: """ Validiert einen Wert mit allen Validatoren. Args: value: Der zu validierende Wert. Raises: ValidationError: Wenn Validierung fehlschlägt. """ # Typkonvertierung falls konfiguriert if self.coerce and value is not None: try: value = self.coerce(value) except Exception as e: raise ValidationError( message=f"Konvertierung fehlgeschlagen: {e}", field=self.name, value=value, code="coercion_error", ) # Alle Validatoren anwenden for validator in self.validators: validator.validate(value, self.name) @dataclass class NestedField(Field): """ Feld für verschachtelte Schemas. Ermöglicht rekursive Validierung von komplexen Strukturen. """ schema: "Schema | None" = None """Das verschachtelte Schema.""" def validate(self, value: Any) -> None: """Validiert den Wert mit dem verschachtelten Schema.""" super().validate(value) if self.schema and value is not None: try: self.schema.validate(value) except ValidationError as e: raise ValidationError( message=f"Verschachtelter Fehler: {e.message}", field=self.name, value=value, code="nested_error", errors=e.errors if hasattr(e, "errors") else [], ) @dataclass class ListField(Field): """ Feld für Listen mit Element-Validierung. Validiert Listen und deren einzelne Elemente. """ item_validators: list[Validator] = dataclass_field(default_factory=list) """Validatoren für jedes Element.""" item_schema: "Schema | None" = None """Schema für komplexe Element-Typen.""" min_items: int | None = None """Minimale Anzahl Elemente.""" max_items: int | None = None """Maximale Anzahl Elemente.""" def validate(self, value: Any) -> None: """Validiert die Liste und deren Elemente.""" super().validate(value) if value is None: return if not isinstance(value, (list, tuple)): raise ValidationError( message="Muss eine Liste sein", field=self.name, value=value, code="type_error", ) # Längen-Prüfung if self.min_items is not None and len(value) < self.min_items: raise ValidationError( message=f"Mindestens {self.min_items} Elemente erforderlich", field=self.name, value=value, code="min_items", ) if self.max_items is not None and len(value) > self.max_items: raise ValidationError( message=f"Höchstens {self.max_items} Elemente erlaubt", field=self.name, value=value, code="max_items", ) # Element-Validierung errors = [] for i, item in enumerate(value): item_field = f"{self.name}[{i}]" # Validatoren für Elemente for validator in self.item_validators: try: validator.validate(item, item_field) except ValidationError as e: errors.append(FieldError( field=item_field, message=e.message, value=item, code=e.code, )) # Schema für komplexe Elemente if self.item_schema: try: self.item_schema.validate(item) except ValidationErrors as e: for err in e.errors: errors.append(FieldError( field=f"{item_field}.{err.field}", message=err.message, value=err.value, code=err.code, )) except ValidationError as e: errors.append(FieldError( field=item_field, message=e.message, value=item, code=e.code, )) if errors: raise ValidationErrors(errors) class Schema: """ Schema für strukturierte Datenvalidierung. Definiert die erwartete Struktur und Validierungsregeln für ein Dictionary oder Objekt. """ def __init__( self, name: str = "", fields: Sequence[Field] | None = None, strict: bool = False, allow_extra: bool = True, ) -> None: """ Initialisiert das Schema. Args: name: Name des Schemas. fields: Liste von Feld-Definitionen. strict: Ob alle Felder erforderlich sind (überschreibt Field.required). allow_extra: Ob zusätzliche Felder erlaubt sind. """ self.name = name self._fields: dict[str, Field] = {} self._aliases: dict[str, str] = {} self.strict = strict self.allow_extra = allow_extra if fields: for f in fields: self.add_field(f) def add_field(self, field: Field) -> "Schema": """ Fügt ein Feld zum Schema hinzu. Args: field: Die Feld-Definition. Returns: Self für Method-Chaining. """ self._fields[field.name] = field if field.alias: self._aliases[field.alias] = field.name return self def field( self, name: str, *validators: Validator, required: bool = True, default: Any = None, description: str = "", ) -> "Schema": """ Fügt ein einfaches Feld hinzu (Builder-Pattern). Args: name: Feldname. *validators: Validatoren für das Feld. required: Ob das Feld erforderlich ist. default: Standardwert. description: Beschreibung. Returns: Self für Method-Chaining. """ self.add_field(Field( name=name, validators=list(validators), required=required, default=default, description=description, )) return self def nested( self, name: str, schema: "Schema", required: bool = True, description: str = "", ) -> "Schema": """ Fügt ein verschachteltes Feld hinzu. Args: name: Feldname. schema: Das verschachtelte Schema. required: Ob das Feld erforderlich ist. description: Beschreibung. Returns: Self für Method-Chaining. """ self.add_field(NestedField( name=name, schema=schema, required=required, description=description, )) return self def list_field( self, name: str, *item_validators: Validator, item_schema: "Schema | None" = None, min_items: int | None = None, max_items: int | None = None, required: bool = True, ) -> "Schema": """ Fügt ein Listen-Feld hinzu. Args: name: Feldname. *item_validators: Validatoren für Elemente. item_schema: Schema für komplexe Elemente. min_items: Minimale Anzahl. max_items: Maximale Anzahl. required: Ob das Feld erforderlich ist. Returns: Self für Method-Chaining. """ self.add_field(ListField( name=name, item_validators=list(item_validators), item_schema=item_schema, min_items=min_items, max_items=max_items, required=required, )) return self def _resolve_field_name(self, key: str) -> str | None: """Löst Aliase auf.""" if key in self._fields: return key return self._aliases.get(key) def validate(self, data: dict[str, Any]) -> dict[str, Any]: """ Validiert ein Dictionary gegen das Schema. Args: data: Die zu validierenden Daten. Returns: Validierte und normalisierte Daten. Raises: ValidationErrors: Wenn Validierung fehlschlägt. """ if not isinstance(data, dict): raise ValidationError( message="Daten müssen ein Dictionary sein", field="", value=data, code="type_error", ) errors: list[FieldError] = [] result: dict[str, Any] = {} # Bekannte Felder verarbeiten for field_name, field_def in self._fields.items(): # Wert aus data holen (mit Alias-Unterstützung) value = None found = False if field_name in data: value = data[field_name] found = True elif field_def.alias and field_def.alias in data: value = data[field_def.alias] found = True # Erforderlichkeit prüfen is_required = self.strict or field_def.required if not found: if is_required: errors.append(FieldError( field=field_name, message="Dieses Feld ist erforderlich", code="required", )) else: result[field_name] = field_def.get_default() continue # Typkonvertierung if field_def.coerce and value is not None: try: value = field_def.coerce(value) except Exception as e: errors.append(FieldError( field=field_name, message=f"Konvertierung fehlgeschlagen: {e}", value=value, code="coercion_error", )) continue # Validierung try: field_def.validate(value) result[field_name] = value except ValidationErrors as e: errors.extend(e.errors) except ValidationError as e: errors.append(FieldError( field=field_name, message=e.message, value=value, code=e.code, )) # Extra-Felder prüfen if not self.allow_extra: known_keys = set(self._fields.keys()) | set(self._aliases.keys()) extra_keys = set(data.keys()) - known_keys for key in extra_keys: errors.append(FieldError( field=key, message="Unbekanntes Feld", value=data[key], code="extra_field", )) else: # Extra-Felder übernehmen known_keys = set(self._fields.keys()) | set(self._aliases.keys()) for key in data: if key not in known_keys: result[key] = data[key] if errors: raise ValidationErrors(errors) return result def is_valid(self, data: dict[str, Any]) -> bool: """ Prüft, ob Daten valide sind. Args: data: Die zu prüfenden Daten. Returns: True wenn valide. """ try: self.validate(data) return True except (ValidationError, ValidationErrors): return False def get_errors(self, data: dict[str, Any]) -> list[FieldError]: """ Gibt alle Validierungsfehler zurück. Args: data: Die zu prüfenden Daten. Returns: Liste von FieldErrors. """ try: self.validate(data) return [] except ValidationErrors as e: return e.errors except ValidationError as e: return e.errors def validate_dict( data: dict[str, Any], schema: Schema, ) -> dict[str, Any]: """ Validiert ein Dictionary mit einem Schema. Args: data: Die zu validierenden Daten. schema: Das Schema für die Validierung. Returns: Validierte Daten. Raises: ValidationErrors: Wenn Validierung fehlschlägt. """ return schema.validate(data) def validate_dataclass( data: dict[str, Any], dataclass_type: Type[T], strict: bool = False, ) -> T: """ Validiert Daten und konvertiert sie in eine Dataclass. Verwendet die Type-Hints der Dataclass für automatische Validierung. Args: data: Die zu validierenden Daten. dataclass_type: Die Ziel-Dataclass. strict: Ob alle Felder erforderlich sind. Returns: Instanz der Dataclass. Raises: ValidationErrors: Wenn Validierung fehlschlägt. """ if not dataclasses.is_dataclass(dataclass_type): raise SchemaError( message="Typ muss eine Dataclass sein", schema_name=dataclass_type.__name__, ) # Schema aus Dataclass generieren schema = Schema(name=dataclass_type.__name__, strict=strict) type_hints = get_type_hints(dataclass_type) for dc_field in dataclasses.fields(dataclass_type): field_type = type_hints.get(dc_field.name, Any) # Erforderlichkeit aus Default ableiten has_default = ( dc_field.default is not dataclasses.MISSING or dc_field.default_factory is not dataclasses.MISSING ) validators: list[Validator] = [] # Typ-Validator hinzufügen if field_type is not Any: try: validators.append(TypeValidator(field_type, allow_none=True)) except Exception: pass # Komplexe Typen ignorieren schema.add_field(Field( name=dc_field.name, validators=validators, required=not has_default, default=( dc_field.default if dc_field.default is not dataclasses.MISSING else None ), default_factory=( dc_field.default_factory if dc_field.default_factory is not dataclasses.MISSING else None ), )) # Validieren validated = schema.validate(data) # Dataclass erstellen return dataclass_type(**validated) # Vordefinierte Schemas für häufige Anwendungsfälle def email_field(name: str = "email", required: bool = True) -> Field: """Erstellt ein E-Mail-Feld mit Pattern-Validierung.""" from trixy_core.utils.validation.validators import PatternValidator return Field( name=name, validators=[ PatternValidator(r"^[^@\s]+@[^@\s]+\.[^@\s]+$"), ], required=required, description="E-Mail-Adresse", ) def url_field(name: str = "url", required: bool = True) -> Field: """Erstellt ein URL-Feld mit Pattern-Validierung.""" from trixy_core.utils.validation.validators import PatternValidator return Field( name=name, validators=[ PatternValidator(r"^https?://[^\s/$.?#].[^\s]*$"), ], required=required, description="URL", ) def phone_field(name: str = "phone", required: bool = True) -> Field: """Erstellt ein Telefon-Feld mit Pattern-Validierung.""" from trixy_core.utils.validation.validators import PatternValidator return Field( name=name, validators=[ PatternValidator(r"^\+?[\d\s\-()]+$"), ], required=required, description="Telefonnummer", )