Trixy7/trixy_core/config/versioning/migration.py

# -*- coding: utf-8 -*-
"""
Konfigurationsmigrationen für Versionsübergänge.

Ermöglicht strukturierte Migrationen zwischen Konfigurationsversionen
mit Rollback-Unterstützung.
"""

from __future__ import annotations

import copy
import json
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any, Callable, Sequence


class MigrationError(Exception):
    """
    Exception bei Migrationsfehlern.
    """

    def __init__(
        self,
        message: str,
        migration_name: str = "",
        from_version: str = "",
        to_version: str = "",
        original_error: Exception | None = None,
    ) -> None:
        """
        Initialisiert den MigrationError.

        Args:
            message: Fehlermeldung.
            migration_name: Name der fehlgeschlagenen Migration.
            from_version: Ausgangsversion.
            to_version: Zielversion.
            original_error: Ursprünglicher Fehler.
        """
        super().__init__(message)
        self.migration_name = migration_name
        self.from_version = from_version
        self.to_version = to_version
        self.original_error = original_error


@dataclass
class MigrationResult:
    """
    Ergebnis einer Migration.
    """

    success: bool
    """Ob die Migration erfolgreich war."""

    from_version: str
    """Ausgangsversion."""

    to_version: str
    """Zielversion."""

    migrations_applied: list[str] = field(default_factory=list)
    """Liste der angewendeten Migrationen."""

    config: dict[str, Any] = field(default_factory=dict)
    """Resultierende Konfiguration."""

    errors: list[str] = field(default_factory=list)
    """Fehlermeldungen falls nicht erfolgreich."""

    backup_path: str | None = None
    """Pfad zum Backup falls erstellt."""

    executed_at: datetime = field(default_factory=datetime.now)
    """Zeitpunkt der Ausführung."""


class ConfigMigration(ABC):
    """
    Abstrakte Basisklasse für Konfigurationsmigrationen.

    Jede Migration definiert einen Übergang von einer Version
    zur nächsten.

    Example:
        class Migration_1_0_to_2_0(ConfigMigration):
            from_version = "1.0"
            to_version = "2.0"

            def up(self, config):
                # Alt: {"server": {"host": "..."}}
                # Neu: {"network": {"host": "..."}}
                if "server" in config:
                    config["network"] = config.pop("server")
                return config

            def down(self, config):
                if "network" in config:
                    config["server"] = config.pop("network")
                return config
    """

    from_version: str = ""
    """Ausgangsversion."""

    to_version: str = ""
    """Zielversion."""

    description: str = ""
    """Beschreibung der Migration."""

    @property
    def name(self) -> str:
        """Name der Migration."""
        return f"{self.from_version}_to_{self.to_version}"

    @abstractmethod
    def up(self, config: dict[str, Any]) -> dict[str, Any]:
        """
        Führt die Migration vorwärts aus.

        Args:
            config: Aktuelle Konfiguration.

        Returns:
            Migrierte Konfiguration.
        """
        pass

    def down(self, config: dict[str, Any]) -> dict[str, Any]:
        """
        Führt die Migration rückwärts aus (Rollback).

        Args:
            config: Aktuelle Konfiguration.

        Returns:
            Zurückmigrierte Konfiguration.

        Note:
            Optional. Nicht alle Migrationen sind umkehrbar.
        """
        raise NotImplementedError(
            f"Migration {self.name} unterstützt kein Rollback"
        )

    def validate_before(self, config: dict[str, Any]) -> bool:
        """
        Validiert die Konfiguration vor der Migration.

        Args:
            config: Zu prüfende Konfiguration.

        Returns:
            True wenn gültig.
        """
        return True

    def validate_after(self, config: dict[str, Any]) -> bool:
        """
        Validiert die Konfiguration nach der Migration.

        Args:
            config: Zu prüfende Konfiguration.

        Returns:
            True wenn gültig.
        """
        return True


class SimpleMigration(ConfigMigration):
    """
    Einfache Migration mit Funktionen statt Klassen.
    """

    def __init__(
        self,
        from_version: str,
        to_version: str,
        up_func: Callable[[dict[str, Any]], dict[str, Any]],
        down_func: Callable[[dict[str, Any]], dict[str, Any]] | None = None,
        description: str = "",
    ) -> None:
        """
        Initialisiert die einfache Migration.

        Args:
            from_version: Ausgangsversion.
            to_version: Zielversion.
            up_func: Vorwärts-Migrationsfunktion.
            down_func: Optionale Rückwärts-Migrationsfunktion.
            description: Beschreibung.
        """
        self.from_version = from_version
        self.to_version = to_version
        self._up_func = up_func
        self._down_func = down_func
        self.description = description

    def up(self, config: dict[str, Any]) -> dict[str, Any]:
        """Führt die Migration vorwärts aus."""
        return self._up_func(config)

    def down(self, config: dict[str, Any]) -> dict[str, Any]:
        """Führt die Migration rückwärts aus."""
        if self._down_func:
            return self._down_func(config)
        raise NotImplementedError(
            f"Migration {self.name} unterstützt kein Rollback"
        )


class ConfigMigrator:
    """
    Verwaltet und führt Konfigurationsmigrationen aus.

    Findet automatisch den Migrationspfad zwischen Versionen
    und führt die notwendigen Migrationen aus.

    Example:
        migrator = ConfigMigrator()

        # Migrationen registrieren
        migrator.register(Migration_1_0_to_2_0())
        migrator.register(Migration_2_0_to_3_0())

        # Migrieren
        result = migrator.migrate(
            config=old_config,
            from_version="1.0",
            to_version="3.0"
        )
    """

    VERSION_KEY = "_config_version"
    """Schlüssel für die Versionsangabe in der Konfiguration."""

    def __init__(
        self,
        version_key: str | None = None,
        backup_dir: str | Path | None = None,
    ) -> None:
        """
        Initialisiert den Migrator.

        Args:
            version_key: Schlüssel für die Versionsangabe.
            backup_dir: Verzeichnis für Backups.
        """
        self._version_key = version_key or self.VERSION_KEY
        self._backup_dir = Path(backup_dir) if backup_dir else None
        self._migrations: dict[str, ConfigMigration] = {}
        self._version_graph: dict[str, dict[str, str]] = {}

    def register(self, migration: ConfigMigration) -> "ConfigMigrator":
        """
        Registriert eine Migration.

        Args:
            migration: Die zu registrierende Migration.

        Returns:
            Self für Method-Chaining.
        """
        key = f"{migration.from_version}:{migration.to_version}"
        self._migrations[key] = migration

        # Graph aktualisieren
        if migration.from_version not in self._version_graph:
            self._version_graph[migration.from_version] = {}
        self._version_graph[migration.from_version][migration.to_version] = key

        return self

    def register_simple(
        self,
        from_version: str,
        to_version: str,
        up_func: Callable[[dict[str, Any]], dict[str, Any]],
        down_func: Callable[[dict[str, Any]], dict[str, Any]] | None = None,
        description: str = "",
    ) -> "ConfigMigrator":
        """
        Registriert eine einfache Migration.

        Args:
            from_version: Ausgangsversion.
            to_version: Zielversion.
            up_func: Vorwärts-Funktion.
            down_func: Rückwärts-Funktion.
            description: Beschreibung.

        Returns:
            Self für Method-Chaining.
        """
        migration = SimpleMigration(
            from_version=from_version,
            to_version=to_version,
            up_func=up_func,
            down_func=down_func,
            description=description,
        )
        return self.register(migration)

    def _find_path(
        self,
        from_version: str,
        to_version: str,
    ) -> list[str] | None:
        """
        Findet den Migrationspfad zwischen zwei Versionen.

        Args:
            from_version: Ausgangsversion.
            to_version: Zielversion.

        Returns:
            Liste von Migrationsschlüsseln oder None.
        """
        if from_version == to_version:
            return []

        # BFS für kürzesten Pfad
        from collections import deque

        queue = deque([(from_version, [])])
        visited = {from_version}

        while queue:
            current, path = queue.popleft()

            if current not in self._version_graph:
                continue

            for next_version, migration_key in self._version_graph[current].items():
                if next_version == to_version:
                    return path + [migration_key]

                if next_version not in visited:
                    visited.add(next_version)
                    queue.append((next_version, path + [migration_key]))

        return None

    def _create_backup(
        self,
        config: dict[str, Any],
        version: str,
    ) -> str | None:
        """Erstellt ein Backup der Konfiguration."""
        if not self._backup_dir:
            return None

        self._backup_dir.mkdir(parents=True, exist_ok=True)
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        backup_path = self._backup_dir / f"config_v{version}_{timestamp}.json"

        with open(backup_path, "w", encoding="utf-8") as f:
            json.dump(config, f, indent=2)

        return str(backup_path)

    def get_version(self, config: dict[str, Any]) -> str | None:
        """
        Ermittelt die Version einer Konfiguration.

        Args:
            config: Die Konfiguration.

        Returns:
            Versionsstring oder None.
        """
        return config.get(self._version_key)

    def set_version(
        self,
        config: dict[str, Any],
        version: str,
    ) -> dict[str, Any]:
        """
        Setzt die Version in einer Konfiguration.

        Args:
            config: Die Konfiguration.
            version: Die zu setzende Version.

        Returns:
            Aktualisierte Konfiguration.
        """
        config = dict(config)
        config[self._version_key] = version
        return config

    def migrate(
        self,
        config: dict[str, Any],
        from_version: str | None = None,
        to_version: str = "",
        create_backup: bool = True,
        dry_run: bool = False,
    ) -> MigrationResult:
        """
        Migriert eine Konfiguration.

        Args:
            config: Die zu migrierende Konfiguration.
            from_version: Ausgangsversion (oder aus config lesen).
            to_version: Zielversion.
            create_backup: Ob ein Backup erstellt werden soll.
            dry_run: Ob nur simuliert werden soll.

        Returns:
            MigrationResult mit Ergebnis.
        """
        # Kopie erstellen
        config = copy.deepcopy(config)

        # Version ermitteln
        if from_version is None:
            from_version = self.get_version(config)
            if from_version is None:
                return MigrationResult(
                    success=False,
                    from_version="",
                    to_version=to_version,
                    errors=["Keine Ausgangsversion gefunden"],
                )

        # Pfad finden
        path = self._find_path(from_version, to_version)
        if path is None:
            return MigrationResult(
                success=False,
                from_version=from_version,
                to_version=to_version,
                errors=[
                    f"Kein Migrationspfad von {from_version} nach {to_version}"
                ],
            )

        if not path:
            return MigrationResult(
                success=True,
                from_version=from_version,
                to_version=to_version,
                config=config,
            )

        # Backup erstellen
        backup_path = None
        if create_backup and not dry_run:
            backup_path = self._create_backup(config, from_version)

        # Migrationen ausführen
        applied: list[str] = []
        errors: list[str] = []
        current_config = config

        for migration_key in path:
            migration = self._migrations[migration_key]

            try:
                # Validierung vor Migration
                if not migration.validate_before(current_config):
                    errors.append(
                        f"Vor-Validierung für {migration.name} fehlgeschlagen"
                    )
                    break

                # Migration ausführen
                if not dry_run:
                    current_config = migration.up(current_config)

                    # Validierung nach Migration
                    if not migration.validate_after(current_config):
                        errors.append(
                            f"Nach-Validierung für {migration.name} fehlgeschlagen"
                        )
                        break

                applied.append(migration.name)

            except Exception as e:
                errors.append(f"Migration {migration.name} fehlgeschlagen: {e}")
                break

        # Version aktualisieren
        if not errors and not dry_run:
            current_config = self.set_version(current_config, to_version)

        return MigrationResult(
            success=len(errors) == 0,
            from_version=from_version,
            to_version=to_version if not errors else "",
            migrations_applied=applied,
            config=current_config,
            errors=errors,
            backup_path=backup_path,
        )

    def rollback(
        self,
        config: dict[str, Any],
        steps: int = 1,
    ) -> MigrationResult:
        """
        Führt ein Rollback durch.

        Args:
            config: Aktuelle Konfiguration.
            steps: Anzahl der Migrationsschritte zurück.

        Returns:
            MigrationResult mit Ergebnis.

        Note:
            Erfordert reversible Migrationen.
        """
        current_version = self.get_version(config)
        if not current_version:
            return MigrationResult(
                success=False,
                from_version="",
                to_version="",
                errors=["Keine Version in Konfiguration gefunden"],
            )

        # Finde vorherige Versionen
        # Dies ist eine vereinfachte Implementierung
        # In Produktion würde man eine Historie führen

        return MigrationResult(
            success=False,
            from_version=current_version,
            to_version="",
            errors=["Rollback nicht implementiert"],
        )

    def list_migrations(self) -> list[dict[str, str]]:
        """
        Listet alle registrierten Migrationen.

        Returns:
            Liste von Migrations-Infos.
        """
        return [
            {
                "name": m.name,
                "from": m.from_version,
                "to": m.to_version,
                "description": m.description,
            }
            for m in self._migrations.values()
        ]

    def get_available_versions(self) -> set[str]:
        """
        Gibt alle bekannten Versionen zurück.

        Returns:
            Set von Versionsstrings.
        """
        versions: set[str] = set()
        for migration in self._migrations.values():
            versions.add(migration.from_version)
            versions.add(migration.to_version)
        return versions