orchestrata_ultimate.core.config.manager

Reference › orchestrata_ultimate › core › config › manager

A layered configuration manager. ConfigurationManager collects ConfigSource records (type, path, format, watch flag, priority) and merges them into one nested dictionary. Source types are FILE, ENV, CONSUL, ETCD and MEMORY; the listing shows file and environment loading, and files can be JSON, YAML or .env (detected from the extension or content, _detect_file_format). add_file and add_env_prefix are the convenience entry points. create_configuration(config_files, env_prefix, watch_files) builds a manager with the prefix ORCHESTRATA_ by default.

load is asynchronous. Sources are ordered by priority, and _merge_config deep-merges each into the result so that the first, highest-priority source wins for scalar values while nested dictionaries are merged recursively. Environment variables with the prefix are turned into nested keys: the prefix is removed, the name is lower-cased and a double underscore separates levels; _convert_value converts the string to a number, boolean or string by its content. get and set read and write dotted paths.

Validation uses validators registered for path patterns with register_validator; validate returns a list of ConfigValidationError (path, message, severity). Defaults check that *.port is an integer from 1 to 65535, *.timeout is a non-negative integer and *.host is a non-empty string. Files marked to watch are followed with watchfiles, and listeners added with add_listener are notified when the configuration changes; shutdown stops the watchers. The module carries @traces annotations for SPEC-001 story US-009. See the core narrative.

At a glance

Class diagram

Dependency diagram

API

Classes

ConfigSourceType (enum)

Bases: Enum · Line 31

Types of configuration sources

ConfigSource (dataclass)

Decorators: traces(spec='001', story='US-009', reqs=['REQ-001-US-009-006']), dataclass · Line 43

Configuration source

ConfigValidationError (dataclass)

Decorators: traces(spec='001', story='US-009', reqs=['REQ-001-US-009-004']), dataclass · Line 55

Configuration validation error

ConfigurationManager

Decorators: traces(spec='001', story='US-009', reqs=['REQ-001-US-009-001', 'REQ-001-US-009-002', 'REQ-001-US-009-004', 'REQ-001-US-… · Line 81

Unified configuration management system.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/config/manager.py

# mypy: ignore-errors
"""
Configuration Management for Orchestrata.

Provides a unified configuration system that supports multiple sources,
validation, and dynamic updates.
"""

import asyncio
import inspect
import json
import logging
import os
import re
from collections.abc import Callable
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Any

import yaml
from watchfiles import Change, awatch

from orchestrata_ultimate.core.base.traces import traces

# ============================================================================
# Data Classes
# ============================================================================


class ConfigSourceType(Enum):
    """Types of configuration sources"""

    FILE = "file"
    ENV = "environment"
    CONSUL = "consul"
    ETCD = "etcd"
    MEMORY = "memory"


@traces(spec="001", story="US-009", reqs=["REQ-001-US-009-006"])
@dataclass
class ConfigSource:
    """Configuration source"""

    type: ConfigSourceType
    path: str
    format: str = "auto"  # "json", "yaml", "env", "auto"
    watch: bool = False
    priority: int = 0  # Higher priority overrides lower


@traces(spec="001", story="US-009", reqs=["REQ-001-US-009-004"])
@dataclass
class ConfigValidationError:
    """Configuration validation error"""

    path: str
    message: str
    severity: str = "error"  # "error", "warning", "info"


# ============================================================================
# Configuration Manager
# ============================================================================


@traces(
    spec="001",
    story="US-009",
    reqs=[
        "REQ-001-US-009-001",  # loads config from YAML files (add_file/_load_file)
        "REQ-001-US-009-002",  # loads config from env vars (add_env_prefix/_load_env)
        "REQ-001-US-009-004",  # validates configuration schema (register_validator/validate)
        "REQ-001-US-009-006",  # priority-based source merging (env > file > defaults)
        "REQ-001-US-009-011",  # reloads config without restart via file watchers
        "REQ-001-US-009-015",  # notifies registered listeners of configuration changes
        "REQ-001-US-009-021",  # exposes the current configuration (get/to_dict)
    ],
)
class ConfigurationManager:
    """
    Unified configuration management system.

    Features:
    - Multiple configuration sources (files, environment, remote)
    - Priority-based merging
    - Validation with schemas
    - Dynamic updates with watchers
    - Type conversion and interpolation
    """

    def __init__(self, name: str = "default") -> None:
        """
        Initialize configuration manager.

        Args:
            name: Manager name for logging
        """
        self.name = name
        self._config: dict[str, Any] = {}
        self._sources: list[ConfigSource] = []
        self._validators: dict[str, Callable] = {}
        self._watchers: dict[str, asyncio.Task] = {}
        self._listeners: list[Callable] = []
        self._logger = logging.getLogger(f"config.{name}")

        # Default validators
        self._register_default_validators()

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-006"])
    def add_source(self, source: ConfigSource) -> None:
        """
        Add configuration source.

        Args:
            source: Configuration source
        """
        self._sources.append(source)
        self._sources.sort(key=lambda s: s.priority, reverse=True)
        self._logger.debug(f"Added config source: {source.type.value}://{source.path}")

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-001"])
    def add_file(self, path: str, watch: bool = False, priority: int = 0) -> None:
        """
        Add configuration file.

        Args:
            path: File path
            watch: Whether to watch for changes
            priority: Source priority
        """
        # Determine format from extension
        format = "auto"
        if path.endswith(".json"):
            format = "json"
        elif path.endswith(".yaml") or path.endswith(".yml"):
            format = "yaml"
        elif path.endswith(".env"):
            format = "env"

        source = ConfigSource(
            type=ConfigSourceType.FILE,
            path=path,
            format=format,
            watch=watch,
            priority=priority,
        )

        self.add_source(source)

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-002"])
    def add_env_prefix(self, prefix: str, priority: int = 10) -> None:
        """
        Add environment variables with prefix.

        Args:
            prefix: Environment variable prefix
            priority: Source priority
        """
        source = ConfigSource(
            type=ConfigSourceType.ENV,
            path=prefix,
            format="env",
            watch=False,
            priority=priority,
        )

        self.add_source(source)

    @traces(
        spec="001", story="US-009", reqs=["REQ-001-US-009-007", "REQ-001-US-009-008"]
    )
    async def load(self) -> None:
        """
        Load configuration from all sources.
        """
        self._logger.info("Loading configuration")

        # Clear existing config
        self._config.clear()

        # Load from each source
        for source in self._sources:
            try:
                config = await self._load_source(source)
                self._merge_config(config, source.priority)
                self._logger.debug(
                    f"Loaded config from {source.type.value}://{source.path}"
                )
            except Exception as e:
                self._logger.error(
                    f"Failed to load config from {source.type.value}://{source.path}: {e}"
                )

        # Validate
        errors = self.validate()
        if errors:
            for error in errors:
                if error.severity == "error":
                    self._logger.error(
                        f"Config validation error at {error.path}: {error.message}"
                    )
                elif error.severity == "warning":
                    self._logger.warning(
                        f"Config validation warning at {error.path}: {error.message}"
                    )

        # Start watchers
        await self._start_watchers()

        # Notify listeners
        await self._notify_listeners()

    @traces(
        spec="001", story="US-009", reqs=["REQ-001-US-009-001", "REQ-001-US-009-002"]
    )
    async def _load_source(self, source: ConfigSource) -> dict[str, Any]:
        """
        Load configuration from source.

        Args:
            source: Configuration source

        Returns:
            Configuration dictionary
        """
        if source.type == ConfigSourceType.FILE:
            return await self._load_file(source)
        elif source.type == ConfigSourceType.ENV:
            return self._load_env(source)
        elif source.type == ConfigSourceType.MEMORY:
            return (
                json.loads(source.path) if isinstance(source.path, str) else source.path
            )
        else:
            raise ValueError(f"Unsupported config source type: {source.type}")

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-001"])
    async def _load_file(self, source: ConfigSource) -> dict[str, Any]:
        """
        Load configuration from file.

        Args:
            source: File source

        Returns:
            Configuration dictionary
        """
        path = Path(source.path)

        if not path.exists():
            self._logger.warning(f"Config file not found: {path}")
            return {}

        format = self._detect_file_format(path, source.format)
        return self._parse_file(path, format)

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-001"])
    def _detect_file_format(self, path: Path, hint: str) -> str:
        """Determine the config file format from extension or content.

        Args:
            path: File path.
            hint: Format hint from the source (may be "auto").

        Returns:
            Resolved format string.
        """
        if hint != "auto":
            return hint
        if path.suffix in [".json"]:
            return "json"
        elif path.suffix in [".yaml", ".yml"]:
            return "yaml"
        elif path.suffix in [".env"]:
            return "env"
        # Detect from content
        content = path.read_text()
        return "json" if content.strip().startswith("{") else "yaml"

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-001"])
    def _parse_file(self, path: Path, format: str) -> dict[str, Any]:
        """Parse a config file according to its format.

        Args:
            path: File path.
            format: Format string ("json", "yaml", or "env").

        Returns:
            Configuration dictionary.
        """
        if format == "json":
            with open(path) as f:
                return json.load(f)
        elif format == "yaml":
            with open(path) as f:
                return yaml.safe_load(f)
        elif format == "env":
            return self._load_env_file(path)
        else:
            raise ValueError(f"Unsupported config format: {format}")

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-002"])
    def _load_env(self, source: ConfigSource) -> dict[str, Any]:
        """
        Load configuration from environment variables.

        Args:
            source: Environment source

        Returns:
            Configuration dictionary
        """
        config = {}
        prefix = source.path

        for key, value in os.environ.items():
            if key.startswith(prefix):
                # Convert key to nested dict path
                config_key = key[len(prefix) :].lstrip("_")
                parts = config_key.lower().split("__")

                # Build nested structure
                current = config
                for i, part in enumerate(parts):
                    if i == len(parts) - 1:
                        # Convert value based on content
                        current[part] = self._convert_value(value)
                    else:
                        if part not in current:
                            current[part] = {}
                        current = current[part]

        return config

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-002"])
    def _load_env_file(self, path: Path) -> dict[str, Any]:
        """
        Load configuration from .env file.

        Args:
            path: .env file path

        Returns:
            Configuration dictionary
        """
        config = {}

        with open(path) as f:
            for line in f:
                line = line.strip()
                if not line or line.startswith("#"):
                    continue

                if "=" in line:
                    key, value = line.split("=", 1)
                    key = key.strip()
                    value = value.strip().strip('"').strip("'")

                    # Convert to nested dict
                    parts = key.lower().split("__")
                    current = config
                    for i, part in enumerate(parts):
                        if i == len(parts) - 1:
                            current[part] = self._convert_value(value)
                        else:
                            if part not in current:
                                current[part] = {}
                            current = current[part]

        return config

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-002"])
    def _convert_value(self, value: str) -> Any:
        """
        Convert string value to appropriate type.

        Args:
            value: String value

        Returns:
            Converted value
        """
        # Try to convert to bool
        if value.lower() in ["true", "yes", "on", "1"]:
            return True
        elif value.lower() in ["false", "no", "off", "0"]:
            return False

        # Try to convert to int
        try:
            return int(value)
        except ValueError as exc:
            logging.getLogger(__name__).debug("int parse: %s", exc)

        # Try to convert to float
        try:
            return float(value)
        except ValueError as exc:
            logging.getLogger(__name__).debug("float parse: %s", exc)

        # Try to convert to list
        if "," in value:
            return [self._convert_value(v.strip()) for v in value.split(",")]

        # Return as string
        return value

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-006"])
    def _merge_config(self, new_config: dict[str, Any], priority: int) -> None:
        """
        Merge configuration with priority.

        Sources are sorted descending by priority before loading (see add_source),
        so each subsequent call to _merge_config applies a lower-priority source.
        _deep_update overwrites existing keys, meaning the first (highest-priority)
        source wins for scalar values while nested dicts are merged recursively.
        This gives correct priority semantics without needing to track priority
        per-key at merge time.

        Args:
            new_config: New configuration to merge
            priority: Priority of this config (informational; ordering is enforced
                      at source-load time by sorting self._sources before iteration)
        """
        self._deep_update(self._config, new_config)

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-006"])
    def _deep_update(self, target: dict[str, Any], source: dict[str, Any]) -> None:
        """
        Deep update dictionary.

        Args:
            target: Target dictionary
            source: Source dictionary
        """
        for key, value in source.items():
            if (
                key in target
                and isinstance(target[key], dict)
                and isinstance(value, dict)
            ):
                self._deep_update(target[key], value)
            else:
                target[key] = value

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-024"])
    def get(self, key: str, default: Any = None) -> Any:
        """
        Get configuration value.

        Args:
            key: Dot-separated key path
            default: Default value if key not found

        Returns:
            Configuration value
        """
        parts = key.split(".")
        current = self._config

        for part in parts:
            if isinstance(current, dict) and part in current:
                current = current[part]
            else:
                return default

        return current

    @traces(
        spec="001", story="US-009", reqs=["REQ-001-US-009-012", "REQ-001-US-009-013"]
    )
    def set(self, key: str, value: Any) -> None:
        """
        Set configuration value.

        Args:
            key: Dot-separated key path
            value: Value to set
        """
        parts = key.split(".")
        current = self._config

        # Navigate to parent
        for part in parts[:-1]:
            if part not in current or not isinstance(current[part], dict):
                current[part] = {}
            current = current[part]

        # Set value
        last_part = parts[-1]
        current[last_part] = value

        # Validate
        errors = self._validate_path(key, value)
        for error in errors:
            self._logger.warning(f"Config validation: {error.message}")

        # Notify listeners
        try:
            loop = asyncio.get_running_loop()
            loop.create_task(self._notify_listeners())
        except RuntimeError:
            pass

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-004"])
    def register_validator(self, path: str, validator: Callable) -> None:
        """
        Register configuration validator.

        Args:
            path: Path pattern to validate (supports * wildcard)
            validator: Validator function that takes (path, value) and returns list of errors
        """
        self._validators[path] = validator

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-004"])
    def validate(self) -> list[ConfigValidationError]:
        """
        Validate entire configuration.

        Returns:
            List of validation errors
        """
        errors = []

        for path_pattern, validator in self._validators.items():
            # Find matching paths
            matching_paths = self._find_matching_paths(path_pattern)

            for path in matching_paths:
                value = self.get(path)
                if value is not None:
                    validator_errors = validator(path, value)
                    if validator_errors:
                        if isinstance(validator_errors, str):
                            errors.append(
                                ConfigValidationError(
                                    path=path, message=validator_errors
                                )
                            )
                        else:
                            for error in validator_errors:
                                if isinstance(error, str):
                                    errors.append(
                                        ConfigValidationError(path=path, message=error)
                                    )
                                else:
                                    errors.append(error)

        return errors

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-013"])
    def _validate_path(self, path: str, value: Any) -> list[ConfigValidationError]:
        """
        Validate specific path.

        Args:
            path: Configuration path
            value: Configuration value

        Returns:
            List of validation errors
        """
        errors = []

        for path_pattern, validator in self._validators.items():
            if self._path_matches_pattern(path, path_pattern):
                validator_errors = validator(path, value)
                if validator_errors:
                    if isinstance(validator_errors, str):
                        errors.append(
                            ConfigValidationError(path=path, message=validator_errors)
                        )
                    else:
                        for error in validator_errors:
                            if isinstance(error, str):
                                errors.append(
                                    ConfigValidationError(path=path, message=error)
                                )
                            else:
                                errors.append(error)

        return errors

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-004"])
    def _find_matching_paths(self, pattern: str) -> list[str]:
        """
        Find configuration paths matching pattern.

        For exact patterns (no wildcards) the lookup is O(1) via get().
        For wildcard patterns, all leaf paths in the config tree are enumerated
        and matched against the compiled regex produced by _path_matches_pattern.

        Limitation: each * matches exactly one path segment (non-greedy).
        A pattern like "*.port" matches "redis.port" but not "a.b.port".
        Use "*.*.port" for two-level nesting.

        Args:
            pattern: Path pattern with * wildcards (each * matches one segment)

        Returns:
            List of matching dot-separated config paths
        """
        if "*" not in pattern:
            return [pattern] if self.get(pattern) is not None else []

        all_paths = self._enumerate_paths(self._config)
        return [p for p in all_paths if self._path_matches_pattern(p, pattern)]

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-004"])
    def _enumerate_paths(self, node: dict[str, Any], prefix: str = "") -> list[str]:
        """Recursively enumerate all dot-separated leaf paths in a config dict.

        Args:
            node: Current dict node.
            prefix: Dot-separated prefix accumulated so far.

        Returns:
            List of full dot-separated paths to every leaf value.
        """
        paths: list[str] = []
        for key, value in node.items():
            full_key = f"{prefix}.{key}" if prefix else key
            if isinstance(value, dict):
                paths.extend(self._enumerate_paths(value, full_key))
            else:
                paths.append(full_key)
        return paths

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-004"])
    def _path_matches_pattern(self, path: str, pattern: str) -> bool:
        """
        Check if path matches pattern.

        Args:
            path: Configuration path
            pattern: Pattern with * wildcards

        Returns:
            True if path matches pattern
        """
        # Simple wildcard matching
        if "*" not in pattern:
            return path == pattern

        # Convert to regex
        regex_pattern = "^" + pattern.replace(".", "\\.").replace("*", "[^.]+") + "$"
        return bool(re.match(regex_pattern, path))

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-004"])
    def _register_default_validators(self) -> None:
        """Register default validators."""

        @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-004"])
        def validate_port(path: str, value: Any) -> str | None:
            if not isinstance(value, int):
                return f"{path} must be an integer"
            if value < 1 or value > 65535:
                return f"{path} must be between 1 and 65535"
            return None

        @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-004"])
        def validate_positive_int(path: str, value: Any) -> str | None:
            if not isinstance(value, int):
                return f"{path} must be an integer"
            if value < 0:
                return f"{path} must be positive"
            return None

        @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-004"])
        def validate_string(path: str, value: Any) -> str | None:
            if not isinstance(value, str):
                return f"{path} must be a string"
            if not value.strip():
                return f"{path} cannot be empty"
            return None

        self.register_validator("*.port", validate_port)
        self.register_validator("*.timeout", validate_positive_int)
        self.register_validator("*.host", validate_string)

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-011"])
    async def _start_watchers(self) -> None:
        """Start file watchers for config sources."""
        for source in self._sources:
            if source.type == ConfigSourceType.FILE and source.watch:
                await self._start_file_watcher(source)

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-011"])
    async def _start_file_watcher(self, source: ConfigSource) -> None:
        """
        Start file watcher for config file.

        Args:
            source: File source to watch
        """
        path = Path(source.path)

        if not path.exists():
            return

        @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-011"])
        async def watch_file() -> None:
            async for changes in awatch(
                path.parent, watch_filter=lambda _, p: p == str(path)
            ):
                for change_type, _ in changes:
                    if change_type in [Change.added, Change.modified]:
                        self._logger.info(f"Config file changed: {path}")
                        await self.load()
                        break

        task = asyncio.create_task(watch_file())
        self._watchers[source.path] = task

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-015"])
    def add_listener(self, listener: Callable) -> None:
        """
        Add configuration change listener.

        Args:
            listener: Callback function that takes (config_manager)
        """
        self._listeners.append(listener)

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-015"])
    async def _notify_listeners(self) -> None:
        """Notify all listeners of configuration changes."""
        for listener in self._listeners:
            try:
                if inspect.iscoroutinefunction(listener):
                    await listener(self)
                else:
                    listener(self)
            except Exception as e:
                self._logger.error(f"Error in config listener: {e}")

    async def shutdown(self) -> None:
        """Shutdown configuration manager."""
        # Cancel watchers
        for task in self._watchers.values():
            task.cancel()

        # Wait for tasks to complete
        if self._watchers:
            await asyncio.gather(*self._watchers.values(), return_exceptions=True)

        self._watchers.clear()
        self._listeners.clear()

        self._logger.info("Configuration manager shutdown")

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-021"])
    def to_dict(self) -> dict[str, Any]:
        """
        Get configuration as dictionary.

        Returns:
            Configuration dictionary
        """
        return self._config.copy()

    @traces(spec="001", story="US-009", reqs=["REQ-001-US-009-026"])
    def __str__(self) -> str:
        """String representation of configuration."""
        return json.dumps(self._config, indent=2, default=str)


# ============================================================================
# Configuration Factory
# ============================================================================


@traces(
    spec="001",
    story="US-009",
    reqs=["REQ-001-US-009-001", "REQ-001-US-009-002", "REQ-001-US-009-006"],
)
def create_configuration(
    config_files: list[str] | None = None,
    env_prefix: str = "ORCHESTRATA_",
    watch_files: bool = False,
) -> ConfigurationManager:
    """
    Create and configure a configuration manager.

    Args:
        config_files: List of config file paths
        env_prefix: Environment variable prefix
        watch_files: Whether to watch config files for changes

    Returns:
        Configured configuration manager
    """
    config = ConfigurationManager()

    # Add default config files
    default_files = [
        "/etc/orchestrata/config.yaml",
        "~/.orchestrata/config.yaml",
        "./config.yaml",
        "./config/config.yaml",
    ]

    files_to_try = (config_files or []) + default_files

    for file_path in files_to_try:
        expanded_path = os.path.expanduser(file_path)
        if os.path.exists(expanded_path):
            config.add_file(expanded_path, watch=watch_files, priority=5)

    # Add environment variables
    config.add_env_prefix(env_prefix, priority=10)

    return config