Trixy7/trixy_core/config/secrets/file_provider.py

# -*- coding: utf-8 -*-
"""
Datei-basierter Secrets-Provider.

Lädt Secrets aus JSON/YAML-Dateien mit optionaler Verschlüsselung.
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any, TypeVar

from trixy_core.config.secrets.provider import SecretsProvider, SecretValue


T = TypeVar("T")


class FileSecretsProvider(SecretsProvider):
    """
    Provider für Secrets aus Dateien.

    Unterstützt JSON und YAML Formate.
    Kann Secrets aus mehreren Dateien laden.

    Example:
        provider = FileSecretsProvider("secrets.json")

        # Aus mehreren Dateien
        provider = FileSecretsProvider([
            "secrets/base.json",
            "secrets/production.json"
        ])

        password = provider.get("database.password")
    """

    def __init__(
        self,
        paths: str | Path | list[str | Path],
        watch: bool = False,
        nested_separator: str = ".",
        file_format: str | None = None,
    ) -> None:
        """
        Initialisiert den File-Provider.

        Args:
            paths: Pfad(e) zu Secret-Dateien.
            watch: Ob Dateien auf Änderungen überwacht werden.
            nested_separator: Trennzeichen für verschachtelte Pfade.
            file_format: Format ('json', 'yaml', oder None für Auto-Detect).
        """
        super().__init__("FileSecretsProvider")

        if isinstance(paths, (str, Path)):
            paths = [paths]

        self._paths = [Path(p) for p in paths]
        self._watch = watch
        self._nested_separator = nested_separator
        self._file_format = file_format
        self._secrets: dict[str, Any] = {}
        self._file_mtimes: dict[Path, float] = {}

        self._load_all()

    def _detect_format(self, path: Path) -> str:
        """Erkennt das Dateiformat anhand der Endung."""
        suffix = path.suffix.lower()
        if suffix in [".yaml", ".yml"]:
            return "yaml"
        return "json"

    def _load_file(self, path: Path) -> dict[str, Any]:
        """Lädt eine einzelne Datei."""
        if not path.exists():
            return {}

        file_format = self._file_format or self._detect_format(path)

        with open(path, "r", encoding="utf-8") as f:
            if file_format == "yaml":
                try:
                    import yaml
                    return yaml.safe_load(f) or {}
                except ImportError:
                    raise ImportError(
                        "PyYAML ist erforderlich für YAML-Dateien"
                    )
            else:
                return json.load(f)

    def _load_all(self) -> None:
        """Lädt alle konfigurierten Dateien."""
        self._secrets.clear()

        for path in self._paths:
            if path.exists():
                data = self._load_file(path)
                self._deep_merge(self._secrets, data)
                self._file_mtimes[path] = path.stat().st_mtime

    def _deep_merge(self, base: dict, override: dict) -> None:
        """Führt zwei Dictionaries tief zusammen."""
        for key, value in override.items():
            if key in base and isinstance(base[key], dict) and isinstance(value, dict):
                self._deep_merge(base[key], value)
            else:
                base[key] = value

    def _get_nested(self, data: dict, path: str) -> Any:
        """Holt einen verschachtelten Wert."""
        keys = path.split(self._nested_separator)
        current = data

        for key in keys:
            if isinstance(current, dict) and key in current:
                current = current[key]
            else:
                return None

        return current

    def _set_nested(self, data: dict, path: str, value: Any) -> None:
        """Setzt einen verschachtelten Wert."""
        keys = path.split(self._nested_separator)
        current = data

        for key in keys[:-1]:
            if key not in current:
                current[key] = {}
            current = current[key]

        current[keys[-1]] = value

    def _flatten(
        self,
        data: dict,
        prefix: str = "",
    ) -> dict[str, Any]:
        """Flacht ein verschachteltes Dictionary ab."""
        result: dict[str, Any] = {}

        for key, value in data.items():
            full_key = f"{prefix}{self._nested_separator}{key}" if prefix else key

            if isinstance(value, dict):
                result.update(self._flatten(value, full_key))
            else:
                result[full_key] = value

        return result

    def _check_reload(self) -> bool:
        """Prüft, ob Dateien neu geladen werden müssen."""
        if not self._watch:
            return False

        for path in self._paths:
            if path.exists():
                mtime = path.stat().st_mtime
                if path not in self._file_mtimes or mtime > self._file_mtimes[path]:
                    return True

        return False

    def get(self, key: str, default: T | None = None) -> T | None:
        """
        Holt einen Secret-Wert.

        Args:
            key: Schlüssel (z.B. 'database.password').
            default: Standardwert.

        Returns:
            Secret-Wert oder Default.
        """
        if self._check_reload():
            self._load_all()

        value = self._get_nested(self._secrets, key)
        return value if value is not None else default

    def get_secret(self, key: str) -> SecretValue | None:
        """
        Holt ein Secret mit Metadaten.

        Args:
            key: Schlüssel.

        Returns:
            SecretValue oder None.
        """
        value = self.get(key)
        if value is None:
            return None

        # Quelle ermitteln
        source_files = [str(p) for p in self._paths if p.exists()]

        return SecretValue(
            key=key,
            value=value,
            source=", ".join(source_files),
            metadata={"paths": source_files},
        )

    def exists(self, key: str) -> bool:
        """Prüft, ob ein Secret existiert."""
        return self.get(key) is not None

    def list_keys(self) -> list[str]:
        """Listet alle Secret-Schlüssel (flach)."""
        if self._check_reload():
            self._load_all()

        return sorted(self._flatten(self._secrets).keys())

    def set(self, key: str, value: Any) -> bool:
        """
        Setzt einen Secret-Wert.

        Speichert in der ersten konfigurierten Datei.

        Args:
            key: Schlüssel.
            value: Wert.

        Returns:
            True wenn erfolgreich.
        """
        self._set_nested(self._secrets, key, value)

        # In erste Datei speichern
        if self._paths:
            primary_path = self._paths[0]

            # Existierende Datei laden und aktualisieren
            if primary_path.exists():
                file_data = self._load_file(primary_path)
            else:
                file_data = {}

            self._set_nested(file_data, key, value)
            self._save_file(primary_path, file_data)

        return True

    def delete(self, key: str) -> bool:
        """
        Löscht ein Secret.

        Args:
            key: Schlüssel.

        Returns:
            True wenn vorhanden.
        """
        keys = key.split(self._nested_separator)
        current = self._secrets

        for k in keys[:-1]:
            if isinstance(current, dict) and k in current:
                current = current[k]
            else:
                return False

        if keys[-1] in current:
            del current[keys[-1]]

            # In Datei speichern
            if self._paths:
                primary_path = self._paths[0]
                if primary_path.exists():
                    file_data = self._load_file(primary_path)

                    # Schlüssel auch aus Datei entfernen
                    try:
                        file_current = file_data
                        for k in keys[:-1]:
                            file_current = file_current[k]
                        if keys[-1] in file_current:
                            del file_current[keys[-1]]
                            self._save_file(primary_path, file_data)
                    except (KeyError, TypeError):
                        pass

            return True

        return False

    def _save_file(self, path: Path, data: dict) -> None:
        """Speichert Daten in eine Datei."""
        path.parent.mkdir(parents=True, exist_ok=True)

        file_format = self._file_format or self._detect_format(path)

        with open(path, "w", encoding="utf-8") as f:
            if file_format == "yaml":
                import yaml
                yaml.dump(data, f, default_flow_style=False)
            else:
                json.dump(data, f, indent=2)

    def reload(self) -> None:
        """Lädt alle Dateien neu."""
        self._load_all()

    def get_all(self) -> dict[str, Any]:
        """
        Gibt alle Secrets als Dictionary zurück.

        Returns:
            Flaches Dictionary mit allen Secrets.
        """
        if self._check_reload():
            self._load_all()

        return self._flatten(self._secrets)

    def get_raw(self) -> dict[str, Any]:
        """
        Gibt die rohe Secrets-Struktur zurück.

        Returns:
            Verschachteltes Dictionary.
        """
        if self._check_reload():
            self._load_all()

        return dict(self._secrets)