# -*- coding: utf-8 -*- """ Umgebungsvariablen-Substitution für Konfigurationen. Ermöglicht die Verwendung von ${VAR} und ${VAR:-default} Syntax in Konfigurationsdateien. """ from __future__ import annotations import os import re from dataclasses import dataclass from typing import Any @dataclass class SubstitutionResult: """ Ergebnis einer Substitution. Enthält den substituierten Wert und Metadaten. """ original: str """Original-String vor Substitution.""" result: str """Ergebnis nach Substitution.""" substitutions: list[tuple[str, str, str]] """Liste von (Platzhalter, Variable, Wert) Tupeln.""" unresolved: list[str] """Liste von Variablen, die nicht aufgelöst werden konnten.""" @property def has_unresolved(self) -> bool: """Prüft, ob unaufgelöste Variablen existieren.""" return len(self.unresolved) > 0 @property def substitution_count(self) -> int: """Anzahl der durchgeführten Substitutionen.""" return len(self.substitutions) class EnvironmentSubstitutor: """ Substituiert Umgebungsvariablen in Strings und Datenstrukturen. Unterstützt folgende Syntax: - ${VAR} - Ersetzt durch Wert von VAR - ${VAR:-default} - Ersetzt durch Wert von VAR oder "default" - ${VAR:=default} - Setzt VAR auf "default" wenn nicht definiert - ${VAR:+value} - Ersetzt durch "value" nur wenn VAR definiert ist - ${VAR:?error} - Wirft Fehler wenn VAR nicht definiert ist Example: substitutor = EnvironmentSubstitutor() # Einfache Substitution result = substitutor.substitute("Host: ${DB_HOST:-localhost}") # Verschachtelte Strukturen config = {"host": "${DB_HOST}", "port": "${DB_PORT:-5432}"} result = substitutor.substitute_recursive(config) """ # Regex-Pattern für Variablen-Syntax PATTERN = re.compile( r"\$\{" r"(?P[A-Za-z_][A-Za-z0-9_]*)" r"(?:" r"(?P:-|:=|:\+|:\?)" r"(?P[^}]*)" r")?" r"\}" ) def __init__( self, env: dict[str, str] | None = None, strict: bool = False, expand_nested: bool = True, ) -> None: """ Initialisiert den Substitutor. Args: env: Optionales Environment-Dictionary (Default: os.environ). strict: Ob bei fehlenden Variablen ein Fehler geworfen wird. expand_nested: Ob Variablen in Default-Werten expandiert werden. """ self._env = env if env is not None else dict(os.environ) self._strict = strict self._expand_nested = expand_nested def get_variable(self, name: str, default: str | None = None) -> str | None: """ Holt eine Umgebungsvariable. Args: name: Variablenname. default: Standardwert falls nicht definiert. Returns: Variablenwert oder Default. """ return self._env.get(name, default) def set_variable(self, name: str, value: str) -> None: """ Setzt eine Umgebungsvariable. Args: name: Variablenname. value: Wert. """ self._env[name] = value def _replace_match( self, match: re.Match, substitutions: list[tuple[str, str, str]], unresolved: list[str], ) -> str: """Ersetzt einen einzelnen Match.""" var_name = match.group("var") operator = match.group("op") default_value = match.group("default") or "" placeholder = match.group(0) value = self._env.get(var_name) if operator is None: # ${VAR} - Einfache Substitution if value is not None: substitutions.append((placeholder, var_name, value)) return value elif self._strict: raise ValueError( f"Umgebungsvariable '{var_name}' ist nicht definiert" ) else: unresolved.append(var_name) return placeholder elif operator == ":-": # ${VAR:-default} - Default wenn nicht gesetzt oder leer if value: substitutions.append((placeholder, var_name, value)) return value else: # Default-Wert ggf. auch expandieren if self._expand_nested: default_value = self.substitute(default_value) substitutions.append((placeholder, var_name, default_value)) return default_value elif operator == ":=": # ${VAR:=default} - Setze Variable wenn nicht definiert if value: substitutions.append((placeholder, var_name, value)) return value else: if self._expand_nested: default_value = self.substitute(default_value) self._env[var_name] = default_value substitutions.append((placeholder, var_name, default_value)) return default_value elif operator == ":+": # ${VAR:+value} - Value nur wenn Variable definiert if value: if self._expand_nested: default_value = self.substitute(default_value) substitutions.append((placeholder, var_name, default_value)) return default_value else: substitutions.append((placeholder, var_name, "")) return "" elif operator == ":?": # ${VAR:?error} - Fehler wenn nicht definiert if value: substitutions.append((placeholder, var_name, value)) return value else: error_msg = default_value or f"Variable '{var_name}' ist nicht definiert" raise ValueError(error_msg) return placeholder def substitute(self, text: str) -> str: """ Substituiert Variablen in einem String. Args: text: String mit Variablenreferenzen. Returns: String mit substituierten Werten. """ result = self.substitute_with_info(text) return result.result def substitute_with_info(self, text: str) -> SubstitutionResult: """ Substituiert Variablen und gibt detaillierte Informationen zurück. Args: text: String mit Variablenreferenzen. Returns: SubstitutionResult mit Details. """ substitutions: list[tuple[str, str, str]] = [] unresolved: list[str] = [] def replacer(match: re.Match) -> str: return self._replace_match(match, substitutions, unresolved) result = self.PATTERN.sub(replacer, text) return SubstitutionResult( original=text, result=result, substitutions=substitutions, unresolved=unresolved, ) def substitute_recursive( self, data: Any, _depth: int = 0, max_depth: int = 100, ) -> Any: """ Substituiert Variablen rekursiv in einer Datenstruktur. Args: data: Datenstruktur (dict, list, str, etc.). _depth: Aktuelle Rekursionstiefe (intern). max_depth: Maximale Rekursionstiefe. Returns: Datenstruktur mit substituierten Werten. """ if _depth > max_depth: raise RecursionError( f"Maximale Rekursionstiefe {max_depth} überschritten" ) if isinstance(data, str): return self.substitute(data) elif isinstance(data, dict): return { self.substitute(k) if isinstance(k, str) else k: self.substitute_recursive( v, _depth + 1, max_depth ) for k, v in data.items() } elif isinstance(data, list): return [ self.substitute_recursive(item, _depth + 1, max_depth) for item in data ] elif isinstance(data, tuple): return tuple( self.substitute_recursive(item, _depth + 1, max_depth) for item in data ) else: return data def list_variables(self, text: str) -> list[str]: """ Listet alle referenzierten Variablen in einem String. Args: text: String mit Variablenreferenzen. Returns: Liste der Variablennamen. """ matches = self.PATTERN.findall(text) return [m[0] for m in matches] # Nur den Variablennamen def list_variables_recursive(self, data: Any) -> set[str]: """ Listet alle Variablen in einer Datenstruktur. Args: data: Datenstruktur. Returns: Set der Variablennamen. """ variables: set[str] = set() if isinstance(data, str): variables.update(self.list_variables(data)) elif isinstance(data, dict): for key, value in data.items(): if isinstance(key, str): variables.update(self.list_variables(key)) variables.update(self.list_variables_recursive(value)) elif isinstance(data, (list, tuple)): for item in data: variables.update(self.list_variables_recursive(item)) return variables def validate(self, data: Any) -> list[str]: """ Validiert, dass alle referenzierten Variablen definiert sind. Args: data: Datenstruktur zu validieren. Returns: Liste der fehlenden Variablen. """ variables = self.list_variables_recursive(data) missing = [var for var in variables if var not in self._env] return missing def substitute_env( text: str, strict: bool = False, ) -> str: """ Convenience-Funktion für Umgebungsvariablen-Substitution. Args: text: String mit Variablenreferenzen. strict: Ob bei fehlenden Variablen ein Fehler geworfen wird. Returns: String mit substituierten Werten. """ substitutor = EnvironmentSubstitutor(strict=strict) return substitutor.substitute(text) def substitute_env_recursive( data: Any, strict: bool = False, ) -> Any: """ Convenience-Funktion für rekursive Substitution. Args: data: Datenstruktur. strict: Ob bei fehlenden Variablen ein Fehler geworfen wird. Returns: Datenstruktur mit substituierten Werten. """ substitutor = EnvironmentSubstitutor(strict=strict) return substitutor.substitute_recursive(data)