geronimo.config.loader

  1"""Configuration loader for Geronimo.
  2
  3Handles loading, parsing, and environment variable interpolation
  4for geronimo.yaml configuration files.
  5"""
  6
  7import os
  8import re
  9from pathlib import Path
 10
 11import yaml
 12from pydantic import ValidationError
 13
 14from geronimo.config.schema import GeronimoConfig
 15
 16
 17class ConfigurationError(Exception):
 18    """Raised when configuration is invalid or cannot be loaded."""
 19
 20    pass
 21
 22
 23def interpolate_env_vars(value: str) -> str:
 24    """Interpolate environment variables in a string.
 25
 26    Supports ${VAR_NAME} and ${VAR_NAME:-default} syntax.
 27
 28    Args:
 29        value: String potentially containing environment variable references.
 30
 31    Returns:
 32        String with environment variables replaced with their values.
 33    """
 34    # Pattern matches ${VAR_NAME} or ${VAR_NAME:-default}
 35    pattern = r"\$\{([A-Z_][A-Z0-9_]*)(?::-([^}]*))?\}"
 36
 37    def replace(match: re.Match) -> str:
 38        var_name = match.group(1)
 39        default = match.group(2)
 40        env_value = os.environ.get(var_name)
 41
 42        if env_value is not None:
 43            return env_value
 44        if default is not None:
 45            return default
 46        # Return original if no value and no default
 47        return match.group(0)
 48
 49    return re.sub(pattern, replace, value)
 50
 51
 52def interpolate_dict(data: dict) -> dict:
 53    """Recursively interpolate environment variables in a dictionary.
 54
 55    Args:
 56        data: Dictionary potentially containing string values with env vars.
 57
 58    Returns:
 59        Dictionary with all string values interpolated.
 60    """
 61    result = {}
 62    for key, value in data.items():
 63        if isinstance(value, str):
 64            result[key] = interpolate_env_vars(value)
 65        elif isinstance(value, dict):
 66            result[key] = interpolate_dict(value)
 67        elif isinstance(value, list):
 68            result[key] = [
 69                interpolate_env_vars(item) if isinstance(item, str) else item
 70                for item in value
 71            ]
 72        else:
 73            result[key] = value
 74    return result
 75
 76
 77def load_config(config_path: str | Path) -> GeronimoConfig:
 78    """Load and validate a geronimo.yaml configuration file.
 79
 80    Args:
 81        config_path: Path to the geronimo.yaml file.
 82
 83    Returns:
 84        Validated GeronimoConfig instance.
 85
 86    Raises:
 87        FileNotFoundError: If the config file doesn't exist.
 88        ConfigurationError: If the config is invalid.
 89    """
 90    path = Path(config_path)
 91
 92    if not path.exists():
 93        raise FileNotFoundError(f"Configuration file not found: {path}")
 94
 95    if not path.is_file():
 96        raise ConfigurationError(f"Path is not a file: {path}")
 97
 98    try:
 99        with open(path) as f:
100            raw_config = yaml.safe_load(f)
101    except yaml.YAMLError as e:
102        raise ConfigurationError(f"Invalid YAML syntax: {e}")
103
104    if raw_config is None:
105        raise ConfigurationError("Configuration file is empty")
106
107    # Interpolate environment variables
108    config_data = interpolate_dict(raw_config)
109
110    # Validate and parse with Pydantic
111    try:
112        return GeronimoConfig(**config_data)
113    except ValidationError as e:
114        errors = []
115        for error in e.errors():
116            location = ".".join(str(loc) for loc in error["loc"])
117            errors.append(f"  {location}: {error['msg']}")
118        raise ConfigurationError(
119            f"Configuration validation failed:\n" + "\n".join(errors)
120        )
121
122
123def save_config(config: GeronimoConfig, config_path: str | Path) -> None:
124    """Save a GeronimoConfig to a YAML file.
125
126    Args:
127        config: The configuration to save.
128        config_path: Path to write the YAML file.
129    """
130    path = Path(config_path)
131    path.parent.mkdir(parents=True, exist_ok=True)
132
133    # Convert to dict, excluding None values for cleaner output
134    config_dict = config.model_dump(exclude_none=True, mode="json")
135
136    with open(path, "w") as f:
137        yaml.dump(config_dict, f, default_flow_style=False, sort_keys=False)