Trixy7/trixy_core/config/secrets/env_provider.py

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

Lädt Secrets aus Umgebungsvariablen mit optionalem Präfix.
"""

from __future__ import annotations

import os
from typing import Any, TypeVar

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


T = TypeVar("T")


class EnvironmentSecretsProvider(SecretsProvider):
    """
    Provider für Secrets aus Umgebungsvariablen.

    Unterstützt Präfixe für Namespacing und verschiedene
    Namenskonventionen.

    Example:
        provider = EnvironmentSecretsProvider(prefix="TRIXY_SECRET_")

        # Liest TRIXY_SECRET_DATABASE_PASSWORD
        password = provider.get("database_password")

        # Mit Konvertierung
        provider = EnvironmentSecretsProvider(
            prefix="APP_",
            transform="upper"  # key wird zu UPPER_CASE
        )
    """

    def __init__(
        self,
        prefix: str = "",
        transform: str = "upper",
        separator: str = "_",
        env: dict[str, str] | None = None,
    ) -> None:
        """
        Initialisiert den Environment-Provider.

        Args:
            prefix: Präfix für Variablennamen.
            transform: Namenstransformation ('upper', 'lower', 'none').
            separator: Trennzeichen für hierarchische Namen.
            env: Optionales Environment-Dictionary.
        """
        super().__init__("EnvironmentSecretsProvider")
        self._prefix = prefix
        self._transform = transform
        self._separator = separator
        self._env = env if env is not None else os.environ

    def _transform_key(self, key: str) -> str:
        """Transformiert einen Schlüssel in den Variablennamen."""
        # Punkte durch Separator ersetzen
        transformed = key.replace(".", self._separator)

        # Transformation anwenden
        if self._transform == "upper":
            transformed = transformed.upper()
        elif self._transform == "lower":
            transformed = transformed.lower()

        # Präfix hinzufügen
        return f"{self._prefix}{transformed}"

    def _reverse_transform_key(self, env_name: str) -> str | None:
        """Transformiert einen Variablennamen zurück in einen Schlüssel."""
        # Präfix entfernen
        if not env_name.startswith(self._prefix):
            return None

        key = env_name[len(self._prefix):]

        # Transformation umkehren
        if self._transform == "upper":
            key = key.lower()
        elif self._transform == "lower":
            key = key.upper()

        # Separator durch Punkte ersetzen
        key = key.replace(self._separator, ".")

        return key

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

        Args:
            key: Schlüssel (wird transformiert).
            default: Standardwert.

        Returns:
            Secret-Wert oder Default.
        """
        env_name = self._transform_key(key)
        return self._env.get(env_name, default)

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

        Args:
            key: Schlüssel.

        Returns:
            SecretValue oder None.
        """
        env_name = self._transform_key(key)
        value = self._env.get(env_name)

        if value is None:
            return None

        return SecretValue(
            key=key,
            value=value,
            source=f"env:{env_name}",
            metadata={"env_name": env_name},
        )

    def exists(self, key: str) -> bool:
        """Prüft, ob die Umgebungsvariable existiert."""
        env_name = self._transform_key(key)
        return env_name in self._env

    def list_keys(self) -> list[str]:
        """Listet alle Schlüssel mit dem konfigurierten Präfix."""
        keys = []
        for env_name in self._env:
            key = self._reverse_transform_key(env_name)
            if key is not None:
                keys.append(key)
        return sorted(keys)

    def set(self, key: str, value: Any) -> bool:
        """
        Setzt eine Umgebungsvariable.

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

        Returns:
            True.
        """
        env_name = self._transform_key(key)
        os.environ[env_name] = str(value)
        if self._env is not os.environ:
            self._env[env_name] = str(value)
        return True

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

        Args:
            key: Schlüssel.

        Returns:
            True wenn vorhanden.
        """
        env_name = self._transform_key(key)
        if env_name in os.environ:
            del os.environ[env_name]
        if env_name in self._env:
            del self._env[env_name]
            return True
        return False

    def get_env_name(self, key: str) -> str:
        """
        Gibt den Umgebungsvariablennamen für einen Schlüssel zurück.

        Args:
            key: Schlüssel.

        Returns:
            Umgebungsvariablenname.
        """
        return self._transform_key(key)

    @classmethod
    def with_dotenv(
        cls,
        dotenv_path: str = ".env",
        prefix: str = "",
        override: bool = False,
    ) -> "EnvironmentSecretsProvider":
        """
        Erstellt einen Provider und lädt eine .env-Datei.

        Args:
            dotenv_path: Pfad zur .env-Datei.
            prefix: Präfix für Variablen.
            override: Ob existierende Variablen überschrieben werden.

        Returns:
            EnvironmentSecretsProvider-Instanz.
        """
        from pathlib import Path

        env = dict(os.environ)
        dotenv = Path(dotenv_path)

        if dotenv.exists():
            with open(dotenv, "r") as f:
                for line in f:
                    line = line.strip()

                    # Kommentare und leere Zeilen überspringen
                    if not line or line.startswith("#"):
                        continue

                    # Name=Wert parsen
                    if "=" in line:
                        name, value = line.split("=", 1)
                        name = name.strip()
                        value = value.strip()

                        # Anführungszeichen entfernen
                        if (
                            (value.startswith('"') and value.endswith('"'))
                            or (value.startswith("'") and value.endswith("'"))
                        ):
                            value = value[1:-1]

                        # Nur setzen wenn nicht vorhanden oder override
                        if override or name not in env:
                            env[name] = value
                            os.environ[name] = value

        return cls(prefix=prefix, env=env)