# -*- coding: utf-8 -*- """ Schema-basierte Validierung für Konfigurationsdateien. Bietet Validierung von Konfigurationen gegen definierte Schemas mit detaillierten Fehlermeldungen. """ from __future__ import annotations import dataclasses from dataclasses import dataclass, field from enum import Enum from typing import Any, Callable, Sequence, Type, get_type_hints class ConfigValueType(Enum): """ Unterstützte Konfigurationswert-Typen. """ STRING = "string" INTEGER = "integer" FLOAT = "float" BOOLEAN = "boolean" LIST = "list" DICT = "dict" ANY = "any" @dataclass class ConfigFieldError: """ Fehler bei der Konfigurationsvalidierung. """ path: str """Pfad zum fehlerhaften Feld (z.B. 'network.host').""" message: str """Fehlermeldung.""" value: Any = None """Der fehlerhafte Wert.""" expected: str = "" """Beschreibung des erwarteten Wertes.""" def __str__(self) -> str: """String-Repräsentation.""" msg = f"{self.path}: {self.message}" if self.expected: msg += f" (erwartet: {self.expected})" return msg class ConfigValidationError(Exception): """ Exception für Konfigurationsvalidierungsfehler. """ def __init__( self, message: str, errors: list[ConfigFieldError] | None = None, ) -> None: """ Initialisiert den Fehler. Args: message: Übergreifende Fehlermeldung. errors: Liste von Feldfehlern. """ super().__init__(message) self.errors = errors or [] def __str__(self) -> str: """String-Repräsentation.""" if not self.errors: return str(self.args[0]) lines = [str(self.args[0])] for error in self.errors: lines.append(f" - {error}") return "\n".join(lines) @dataclass class ConfigFieldSpec: """ Spezifikation eines Konfigurationsfeldes. """ name: str """Feldname.""" value_type: ConfigValueType = ConfigValueType.ANY """Erwarteter Werttyp.""" required: bool = True """Ob das Feld erforderlich ist.""" default: Any = None """Standardwert.""" min_value: float | None = None """Minimaler Wert (für Zahlen).""" max_value: float | None = None """Maximaler Wert (für Zahlen).""" min_length: int | None = None """Minimale Länge (für Strings/Listen).""" max_length: int | None = None """Maximale Länge (für Strings/Listen).""" pattern: str | None = None """Regex-Pattern (für Strings).""" choices: Sequence[Any] | None = None """Erlaubte Werte.""" validator: Callable[[Any], bool] | None = None """Benutzerdefinierte Validierungsfunktion.""" description: str = "" """Beschreibung für Dokumentation.""" nested_schema: "ConfigSchema | None" = None """Schema für verschachtelte Dicts.""" item_spec: "ConfigFieldSpec | None" = None """Spezifikation für Listen-Elemente.""" class ConfigSchema: """ Schema für Konfigurationsvalidierung. Definiert die erwartete Struktur und Regeln für eine Konfiguration. Example: schema = ConfigSchema("ServerConfig") schema.field("host", ConfigValueType.STRING, required=True) schema.field("port", ConfigValueType.INTEGER, min_value=1, max_value=65535) schema.field("debug", ConfigValueType.BOOLEAN, default=False) # Validieren schema.validate(config_data) """ def __init__( self, name: str = "", strict: bool = False, ) -> None: """ Initialisiert das Schema. Args: name: Name des Schemas. strict: Ob unbekannte Felder einen Fehler verursachen. """ self.name = name self.strict = strict self._fields: dict[str, ConfigFieldSpec] = {} def field( self, name: str, value_type: ConfigValueType = ConfigValueType.ANY, required: bool = True, default: Any = None, min_value: float | None = None, max_value: float | None = None, min_length: int | None = None, max_length: int | None = None, pattern: str | None = None, choices: Sequence[Any] | None = None, validator: Callable[[Any], bool] | None = None, description: str = "", nested_schema: "ConfigSchema | None" = None, item_spec: ConfigFieldSpec | None = None, ) -> "ConfigSchema": """ Fügt eine Feldspezifikation hinzu. Args: name: Feldname. value_type: Erwarteter Typ. required: Ob erforderlich. default: Standardwert. min_value: Minimaler Wert. max_value: Maximaler Wert. min_length: Minimale Länge. max_length: Maximale Länge. pattern: Regex-Pattern. choices: Erlaubte Werte. validator: Benutzerdefinierte Prüffunktion. description: Beschreibung. nested_schema: Schema für verschachtelte Dicts. item_spec: Spezifikation für Listen-Elemente. Returns: Self für Method-Chaining. """ self._fields[name] = ConfigFieldSpec( name=name, value_type=value_type, required=required, default=default, min_value=min_value, max_value=max_value, min_length=min_length, max_length=max_length, pattern=pattern, choices=choices, validator=validator, description=description, nested_schema=nested_schema, item_spec=item_spec, ) return self def nested( self, name: str, schema: "ConfigSchema", required: bool = True, description: str = "", ) -> "ConfigSchema": """ Fügt ein verschachteltes Schema hinzu. Args: name: Feldname. schema: Das verschachtelte Schema. required: Ob erforderlich. description: Beschreibung. Returns: Self für Method-Chaining. """ self._fields[name] = ConfigFieldSpec( name=name, value_type=ConfigValueType.DICT, required=required, description=description, nested_schema=schema, ) return self def _validate_type( self, value: Any, spec: ConfigFieldSpec, path: str, ) -> list[ConfigFieldError]: """Validiert den Typ eines Wertes.""" errors: list[ConfigFieldError] = [] if spec.value_type == ConfigValueType.ANY: return errors type_map = { ConfigValueType.STRING: str, ConfigValueType.INTEGER: int, ConfigValueType.FLOAT: (int, float), ConfigValueType.BOOLEAN: bool, ConfigValueType.LIST: (list, tuple), ConfigValueType.DICT: dict, } expected_type = type_map.get(spec.value_type) if expected_type and not isinstance(value, expected_type): errors.append(ConfigFieldError( path=path, message=f"Ungültiger Typ: {type(value).__name__}", value=value, expected=spec.value_type.value, )) return errors def _validate_constraints( self, value: Any, spec: ConfigFieldSpec, path: str, ) -> list[ConfigFieldError]: """Validiert Constraints für einen Wert.""" errors: list[ConfigFieldError] = [] # Bereichsprüfung if spec.min_value is not None and isinstance(value, (int, float)): if value < spec.min_value: errors.append(ConfigFieldError( path=path, message=f"Wert {value} ist kleiner als Minimum", value=value, expected=f">= {spec.min_value}", )) if spec.max_value is not None and isinstance(value, (int, float)): if value > spec.max_value: errors.append(ConfigFieldError( path=path, message=f"Wert {value} ist größer als Maximum", value=value, expected=f"<= {spec.max_value}", )) # Längenprüfung if hasattr(value, "__len__"): if spec.min_length is not None and len(value) < spec.min_length: errors.append(ConfigFieldError( path=path, message=f"Länge {len(value)} ist zu kurz", value=value, expected=f">= {spec.min_length}", )) if spec.max_length is not None and len(value) > spec.max_length: errors.append(ConfigFieldError( path=path, message=f"Länge {len(value)} ist zu lang", value=value, expected=f"<= {spec.max_length}", )) # Pattern-Prüfung if spec.pattern and isinstance(value, str): import re if not re.match(spec.pattern, value): errors.append(ConfigFieldError( path=path, message="Wert entspricht nicht dem erwarteten Muster", value=value, expected=f"Pattern: {spec.pattern}", )) # Choices-Prüfung if spec.choices is not None: if value not in spec.choices: errors.append(ConfigFieldError( path=path, message=f"Ungültiger Wert", value=value, expected=f"einer von {list(spec.choices)}", )) # Custom Validator if spec.validator: try: if not spec.validator(value): errors.append(ConfigFieldError( path=path, message="Benutzerdefinierte Validierung fehlgeschlagen", value=value, )) except Exception as e: errors.append(ConfigFieldError( path=path, message=f"Validierungsfehler: {e}", value=value, )) return errors def _validate_field( self, data: dict, spec: ConfigFieldSpec, path_prefix: str, ) -> list[ConfigFieldError]: """Validiert ein einzelnes Feld.""" errors: list[ConfigFieldError] = [] path = f"{path_prefix}.{spec.name}" if path_prefix else spec.name # Existenz prüfen if spec.name not in data: if spec.required: errors.append(ConfigFieldError( path=path, message="Erforderliches Feld fehlt", )) return errors value = data[spec.name] # None-Wert if value is None: if spec.required: errors.append(ConfigFieldError( path=path, message="Erforderliches Feld darf nicht null sein", value=value, )) return errors # Typ validieren errors.extend(self._validate_type(value, spec, path)) # Constraints validieren errors.extend(self._validate_constraints(value, spec, path)) # Verschachteltes Schema if spec.nested_schema and isinstance(value, dict): errors.extend(spec.nested_schema.validate_and_collect(value, path)) # Listen-Elemente if spec.item_spec and isinstance(value, (list, tuple)): for i, item in enumerate(value): item_path = f"{path}[{i}]" errors.extend(self._validate_type(item, spec.item_spec, item_path)) errors.extend(self._validate_constraints(item, spec.item_spec, item_path)) if spec.item_spec.nested_schema and isinstance(item, dict): errors.extend( spec.item_spec.nested_schema.validate_and_collect(item, item_path) ) return errors def validate_and_collect( self, data: dict, path_prefix: str = "", ) -> list[ConfigFieldError]: """ Validiert Daten und sammelt alle Fehler. Args: data: Zu validierende Daten. path_prefix: Pfad-Präfix für Fehler. Returns: Liste von Fehlern. """ errors: list[ConfigFieldError] = [] if not isinstance(data, dict): errors.append(ConfigFieldError( path=path_prefix or "(root)", message="Konfiguration muss ein Dictionary sein", value=data, )) return errors # Bekannte Felder validieren for spec in self._fields.values(): errors.extend(self._validate_field(data, spec, path_prefix)) # Unbekannte Felder prüfen if self.strict: known_fields = set(self._fields.keys()) for key in data: if key not in known_fields: path = f"{path_prefix}.{key}" if path_prefix else key errors.append(ConfigFieldError( path=path, message="Unbekanntes Feld", value=data[key], )) return errors def validate(self, data: dict) -> dict: """ Validiert Daten und gibt sie mit Defaults zurück. Args: data: Zu validierende Daten. Returns: Validierte Daten mit Defaults. Raises: ConfigValidationError: Wenn Validierung fehlschlägt. """ errors = self.validate_and_collect(data) if errors: raise ConfigValidationError( f"Konfigurationsvalidierung für '{self.name}' fehlgeschlagen", errors=errors, ) # Defaults anwenden result = dict(data) for name, spec in self._fields.items(): if name not in result and spec.default is not None: result[name] = spec.default return result def is_valid(self, data: dict) -> bool: """ Prüft, ob Daten valide sind. Args: data: Zu prüfende Daten. Returns: True wenn valide. """ errors = self.validate_and_collect(data) return len(errors) == 0 def validate_config( data: dict, schema: ConfigSchema, ) -> dict: """ Validiert eine Konfiguration gegen ein Schema. Args: data: Konfigurationsdaten. schema: Validierungsschema. Returns: Validierte Daten mit Defaults. Raises: ConfigValidationError: Wenn Validierung fehlschlägt. """ return schema.validate(data) def schema_from_dataclass( dataclass_type: Type, strict: bool = False, ) -> ConfigSchema: """ Erstellt ein Schema aus einer Dataclass. Args: dataclass_type: Die Dataclass. strict: Ob unbekannte Felder einen Fehler verursachen. Returns: ConfigSchema basierend auf der Dataclass. """ if not dataclasses.is_dataclass(dataclass_type): raise ValueError("Typ muss eine Dataclass sein") schema = ConfigSchema(name=dataclass_type.__name__, strict=strict) type_hints = get_type_hints(dataclass_type) type_map = { str: ConfigValueType.STRING, int: ConfigValueType.INTEGER, float: ConfigValueType.FLOAT, bool: ConfigValueType.BOOLEAN, list: ConfigValueType.LIST, dict: ConfigValueType.DICT, } for dc_field in dataclasses.fields(dataclass_type): field_type = type_hints.get(dc_field.name, Any) has_default = ( dc_field.default is not dataclasses.MISSING or dc_field.default_factory is not dataclasses.MISSING ) value_type = type_map.get(field_type, ConfigValueType.ANY) default = None if dc_field.default is not dataclasses.MISSING: default = dc_field.default elif dc_field.default_factory is not dataclasses.MISSING: default = dc_field.default_factory() schema.field( name=dc_field.name, value_type=value_type, required=not has_default, default=default, ) return schema