Trixy7/trixy_core/utils/validation/schemas.py

# -*- 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",
    )