# -*- coding: utf-8 -*- """ Konfigurationsvererbung für mehrschichtige Konfigurationen. Ermöglicht das Laden von Basis-Konfigurationen mit Environment-spezifischen Overrides. """ from __future__ import annotations import json from dataclasses import dataclass, field from pathlib import Path from typing import Any, Sequence @dataclass class ConfigLayer: """ Eine Schicht in der Konfigurationshierarchie. """ name: str """Name der Schicht (z.B. 'base', 'production').""" data: dict[str, Any] """Konfigurationsdaten dieser Schicht.""" source: str = "" """Quelle der Konfiguration (z.B. Dateipfad).""" priority: int = 0 """Priorität (höher = überschreibt niedrigere).""" def deep_merge( base: dict[str, Any], override: dict[str, Any], merge_lists: bool = False, ) -> dict[str, Any]: """ Führt zwei Dictionaries tief zusammen. Args: base: Basis-Dictionary. override: Override-Dictionary. merge_lists: Ob Listen zusammengeführt statt ersetzt werden. Returns: Zusammengeführtes Dictionary. """ result = dict(base) for key, value in override.items(): if key in result: base_value = result[key] # Rekursives Mergen für verschachtelte Dicts if isinstance(base_value, dict) and isinstance(value, dict): result[key] = deep_merge(base_value, value, merge_lists) # Optionales Listen-Merging elif merge_lists and isinstance(base_value, list) and isinstance(value, list): result[key] = base_value + value # Sonst ersetzen else: result[key] = value else: result[key] = value return result class ConfigInheritance: """ Verwaltet mehrschichtige Konfigurationen mit Vererbung. Lädt Konfigurationen in der Reihenfolge: base.json → {environment}.json → local.json Spätere Schichten überschreiben frühere. Example: inheritance = ConfigInheritance("config/") config = inheritance.load(environment="production") # Lädt: # 1. config/base.json # 2. config/production.json # 3. config/local.json (falls vorhanden) """ DEFAULT_LAYERS = ["base", "local"] def __init__( self, config_dir: str | Path, extension: str = ".json", merge_lists: bool = False, ) -> None: """ Initialisiert die Konfigurationsvererbung. Args: config_dir: Verzeichnis mit Konfigurationsdateien. extension: Dateiendung (z.B. '.json', '.yaml'). merge_lists: Ob Listen zusammengeführt werden. """ self._config_dir = Path(config_dir) self._extension = extension self._merge_lists = merge_lists self._layers: list[ConfigLayer] = [] self._merged: dict[str, Any] | None = None @property def layers(self) -> list[ConfigLayer]: """Gibt die geladenen Schichten zurück.""" return list(self._layers) @property def merged_config(self) -> dict[str, Any]: """Gibt die zusammengeführte Konfiguration zurück.""" if self._merged is None: self._merged = self._merge_all() return dict(self._merged) def _load_file(self, path: Path) -> dict[str, Any]: """Lädt eine Konfigurationsdatei.""" if not path.exists(): return {} with open(path, "r", encoding="utf-8") as f: if self._extension in [".json", ".jsn"]: return json.load(f) elif self._extension in [".yaml", ".yml"]: try: import yaml return yaml.safe_load(f) or {} except ImportError: raise ImportError( "PyYAML ist erforderlich für YAML-Dateien" ) else: # Versuche JSON als Fallback return json.load(f) def _get_config_path(self, name: str) -> Path: """Erstellt den Pfad zu einer Konfigurationsdatei.""" return self._config_dir / f"{name}{self._extension}" def add_layer( self, name: str, data: dict[str, Any] | None = None, source: str = "", priority: int | None = None, ) -> "ConfigInheritance": """ Fügt eine Konfigurationsschicht hinzu. Args: name: Name der Schicht. data: Konfigurationsdaten (oder None zum Laden aus Datei). source: Quelle der Konfiguration. priority: Priorität (Default: nächsthöher). Returns: Self für Method-Chaining. """ if data is None: path = self._get_config_path(name) data = self._load_file(path) source = str(path) if priority is None: priority = len(self._layers) self._layers.append(ConfigLayer( name=name, data=data, source=source, priority=priority, )) self._merged = None # Cache invalidieren return self def load( self, environment: str | None = None, extra_layers: Sequence[str] | None = None, ) -> dict[str, Any]: """ Lädt und mergt Konfigurationen. Args: environment: Optionale Umgebung (z.B. 'production'). extra_layers: Zusätzliche Schichten. Returns: Zusammengeführte Konfiguration. """ self._layers.clear() # Base laden self.add_layer("base") # Environment laden if environment: self.add_layer(environment) # Extra-Schichten if extra_layers: for layer in extra_layers: self.add_layer(layer) # Local (optional, überschreibt alles) local_path = self._get_config_path("local") if local_path.exists(): self.add_layer("local") return self.merged_config def _merge_all(self) -> dict[str, Any]: """Führt alle Schichten zusammen.""" # Nach Priorität sortieren sorted_layers = sorted(self._layers, key=lambda l: l.priority) result: dict[str, Any] = {} for layer in sorted_layers: result = deep_merge(result, layer.data, self._merge_lists) return result def get_layer_value( self, key: str, layer_name: str | None = None, ) -> tuple[Any, str | None]: """ Findet einen Wert und seine Quelle. Args: key: Konfigurationsschlüssel (z.B. 'network.host'). layer_name: Optionaler spezifischer Layer. Returns: Tupel (Wert, Layer-Name). """ keys = key.split(".") def get_nested(data: dict, keys: list[str]) -> Any: for k in keys: if isinstance(data, dict) and k in data: data = data[k] else: return None return data if layer_name: for layer in self._layers: if layer.name == layer_name: value = get_nested(layer.data, keys) if value is not None: return value, layer.name return None, None # Rückwärts suchen (höchste Priorität zuerst) sorted_layers = sorted( self._layers, key=lambda l: l.priority, reverse=True ) for layer in sorted_layers: value = get_nested(layer.data, keys) if value is not None: return value, layer.name return None, None def get_overrides(self, key: str) -> list[tuple[str, Any]]: """ Findet alle Overrides für einen Schlüssel. Args: key: Konfigurationsschlüssel. Returns: Liste von (Layer-Name, Wert) Tupeln. """ keys = key.split(".") results: list[tuple[str, Any]] = [] def get_nested(data: dict, keys: list[str]) -> Any: for k in keys: if isinstance(data, dict) and k in data: data = data[k] else: return None return data for layer in sorted(self._layers, key=lambda l: l.priority): value = get_nested(layer.data, keys) if value is not None: results.append((layer.name, value)) return results def reload(self) -> dict[str, Any]: """ Lädt alle Schichten neu. Returns: Neu gemergte Konfiguration. """ for layer in self._layers: if layer.source: path = Path(layer.source) if path.exists(): layer.data = self._load_file(path) self._merged = None return self.merged_config class EnvironmentConfig: """ Environment-aware Konfigurationslader. Lädt automatisch die richtige Konfiguration basierend auf Umgebungsvariablen. """ ENV_VAR = "TRIXY_ENV" DEFAULT_ENV = "development" def __init__( self, config_dir: str | Path, env_var: str | None = None, ) -> None: """ Initialisiert den Environment-Konfigurationslader. Args: config_dir: Konfigurationsverzeichnis. env_var: Umgebungsvariable für Environment-Name. """ self._config_dir = Path(config_dir) self._env_var = env_var or self.ENV_VAR self._inheritance = ConfigInheritance(config_dir) @property def environment(self) -> str: """Gibt das aktuelle Environment zurück.""" import os return os.environ.get(self._env_var, self.DEFAULT_ENV) def load( self, environment: str | None = None, ) -> dict[str, Any]: """ Lädt die Konfiguration für ein Environment. Args: environment: Optionales Environment (Default: aus Umgebungsvariable). Returns: Zusammengeführte Konfiguration. """ env = environment or self.environment return self._inheritance.load(environment=env) def get(self, key: str, default: Any = None) -> Any: """ Holt einen Konfigurationswert. Args: key: Konfigurationsschlüssel. default: Standardwert. Returns: Konfigurationswert oder Default. """ config = self._inheritance.merged_config keys = key.split(".") for k in keys: if isinstance(config, dict) and k in config: config = config[k] else: return default return config