orchestrata_ultimate.postman.config

Reference › orchestrata_ultimate › postman › config

The configuration manager of the broker, SPEC-002 stories FR-011 and US-009. PostmanConfigManager implements IPostmanConfig from interfaces and builds a PostmanConfig (defined in types) from a YAML file and from environment variables. Defaults come from constants, and errors use InvalidConfigError from exceptions. No other module imports it.

load takes the YAML file's contents, applies the environment overrides, converts the dictionary into a PostmanConfig, validates it with validate and returns it; a missing YAML library, a parse error, bad values or failed validation raise InvalidConfigError. _apply_env_overrides reads every variable with the prefix POSTMAN_ (the default), lower-cases the rest, treats a double underscore as nesting and converts the value type with _convert_env_value. reload repeats the load under a lock, notifies the on_change callbacks with the changes and returns a success flag and an error message instead of raising. With watch_for_changes set, _start_watch starts a background thread that checks the file's modification time at an interval (2 seconds by default) and reloads, and stop_watch stops it. get reads a value by dotted key.

At a glance

Class diagram

Dependency diagram

API

Classes

PostmanConfigManager

Bases: IPostmanConfig · Decorators: traces(spec='002', story='FR-011/US-009', reqs=['REQ-002-FR11-001', 'REQ-002-FR11-002', 'REQ-002-FR11-005', 'REQ-002-FR… · Line 56

Configuration manager implementing IPostmanConfig.

Source

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

"""
Postman Configuration Manager.

Implements IPostmanConfig interface with support for YAML configuration files,
environment variables, and hot-reload capability.
"""

import dataclasses
import logging
import os
import threading
from collections.abc import Callable
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.postman.constants import (
    CONFIG_WATCH_THREAD_JOIN_TIMEOUT_SECONDS,
)

logger = logging.getLogger(__name__)

# Try to import required third-party libraries
try:
    import yaml

    YAML_AVAILABLE = True
except ImportError:
    yaml = None  # type: ignore[assignment]
    YAML_AVAILABLE = False

# Import Postman types and exceptions with fallback for different project layouts
try:
    from orchestrata_ultimate.postman.exceptions import InvalidConfigError
    from orchestrata_ultimate.postman.interfaces import IPostmanConfig
    from orchestrata_ultimate.postman.types import PostmanConfig
except ImportError:
    # fallback to relative imports if the module is in the same package
    from orchestrata_ultimate.postman.exceptions import InvalidConfigError
    from orchestrata_ultimate.postman.interfaces import IPostmanConfig
    from orchestrata_ultimate.postman.types import PostmanConfig


@traces(
    spec="002",
    story="FR-011/US-009",
    reqs=[
        "REQ-002-FR11-001",
        "REQ-002-FR11-002",
        "REQ-002-FR11-005",
        "REQ-002-FR11-011",
        "REQ-002-US-009-025",
        "REQ-002-US-009-028",
    ],
)
class PostmanConfigManager(IPostmanConfig):
    """Configuration manager implementing IPostmanConfig."""

    def __init__(
        self,
        config_path: str | None = None,
        env_prefix: str = "POSTMAN_",
        watch_for_changes: bool = False,
        watch_interval_seconds: float = 2.0,
        environ: dict[str, str] | None = None,
    ) -> None:
        """
        Initialize configuration manager.

        Args:
            config_path: Path to YAML configuration file.
            env_prefix: Prefix for environment variable overrides.
            watch_for_changes: If True, starts a background thread that checks
                for configuration file changes and automatically reloads.
            watch_interval_seconds: How often to poll for file changes (default 2s).
            environ: Optional explicit environment dict. Defaults to ``os.environ``.
        """
        self._config_path = Path(config_path) if config_path else None
        self._env_prefix = env_prefix
        self._config: PostmanConfig | None = None
        self._callbacks: list[Callable[[str, Any, Any], None]] = []
        self._last_mtime: float = 0.0
        self._lock = threading.RLock()
        self._watch_thread: threading.Thread | None = None
        self._stop_watch = threading.Event()
        self._watch_interval_seconds = watch_interval_seconds
        self._environ = environ

        if watch_for_changes:
            self._start_watch()

        # Load initial configuration
        self.load()

    @traces(
        spec="002",
        story="FR-011/US-009",
        reqs=[
            "REQ-002-FR11-001",
            "REQ-002-FR11-002",
            "REQ-002-FR11-004",
            "REQ-002-FR11-005",
            "REQ-002-FR11-009",
            "REQ-002-US-009-005",
        ],
    )
    def load(self) -> PostmanConfig:
        """
        Load configuration from all sources.

        Returns:
            Merged PostmanConfig instance.

        Raises:
            InvalidConfigError: If the configuration is invalid.
        """
        with self._lock:
            config_dict: dict[str, Any] = {}

            # 1. Load YAML file if provided and available
            if self._config_path and self._config_path.exists():
                if not YAML_AVAILABLE or yaml is None:
                    raise InvalidConfigError(
                        "YAML library not installed. Install PyYAML to use YAML configuration."
                    )
                try:
                    with open(self._config_path) as f:
                        file_config = yaml.safe_load(f) or {}
                    config_dict.update(file_config)
                    self._last_mtime = self._config_path.stat().st_mtime
                except (yaml.YAMLError, OSError) as e:
                    raise InvalidConfigError(f"Failed to load YAML config: {e}")

            # 2. Apply environment variable overrides
            self._apply_env_overrides(config_dict)

            # 3. Convert dictionary to PostmanConfig, filling missing values with defaults
            try:
                self._config = self._dict_to_config(config_dict)
            except (TypeError, ValueError) as e:
                raise InvalidConfigError(f"Invalid configuration values: {e}")

            # 4. Validate
            is_valid, errors = self.validate(config_dict)
            if not is_valid:
                raise InvalidConfigError(f"Configuration validation failed: {errors}")

            return self._config

    @traces(
        spec="002",
        story="US-009",
        reqs=["REQ-002-US-009-025", "REQ-002-US-009-028"],
    )
    def reload(self) -> tuple[bool, str | None]:
        """
        Reload configuration from source files.

        Returns:
            Tuple of (success, error_message)
        """
        try:
            with self._lock:
                old_config = self._config
                new_config = self.load()
                # Notify callbacks about changed values
                if old_config is not None:
                    self._notify_callbacks(old_config, new_config)
                return True, None
        except Exception as e:
            return False, str(e)

    @traces(spec="002", story="FR-011", reqs=["REQ-002-FR11-002"])
    def _apply_env_overrides(self, config_dict: dict[str, Any]) -> None:
        """Apply environment variable overrides to config_dict."""
        env = self._environ if self._environ is not None else os.environ
        for env_key, env_val in env.items():
            if not env_key.startswith(self._env_prefix):
                continue
            # Remove prefix and convert to lowercase
            key = env_key[len(self._env_prefix) :].lower()
            # Support nested keys via double underscore (e.g., POSTMAN_RETRY_POLICY__MAX_RETRIES)
            key_parts = key.split("__")
            target = config_dict
            for part in key_parts[:-1]:
                if part not in target:
                    target[part] = {}
                target = target[part]
            leaf_key = key_parts[-1]
            # Try to convert to appropriate type (int, float, bool, keep string)
            converted = self._convert_env_value(env_val)
            target[leaf_key] = converted

    @staticmethod
    @traces(spec="002", story="FR-011", reqs=["REQ-002-FR11-002"])
    def _convert_env_value(value: str) -> Any:
        """Convert environment variable string to appropriate type."""
        # Try int
        try:
            return int(value)
        except ValueError as exc:
            logger.debug("int parse: %s", exc)
        # Try float
        try:
            return float(value)
        except ValueError as exc:
            logger.debug("float parse: %s", exc)
        # Try bool (case-insensitive true/false)
        if value.lower() in ("true", "false"):
            return value.lower() == "true"
        # Keep as string
        return value

    @traces(spec="002", story="FR-011", reqs=["REQ-002-FR11-009"])
    def _dict_to_config(self, config_dict: dict[str, Any]) -> PostmanConfig:
        """Convert a dictionary to a PostmanConfig instance."""
        # Ensure retry_policy is a dict
        retry_policy = config_dict.get("retry_policy")
        if retry_policy is None:
            retry_policy = PostmanConfig().retry_policy
        elif not isinstance(retry_policy, dict):
            raise ValueError("retry_policy must be a dictionary")

        # Build kwargs with defaults from an empty PostmanConfig
        default_config = PostmanConfig()
        kwargs = default_config.__dict__.copy()
        kwargs.update(config_dict)
        kwargs["retry_policy"] = retry_policy

        # Filter out any extra keys that PostmanConfig doesn't expect
        allowed_keys = {f.name for f in dataclasses.fields(PostmanConfig)}
        filtered_kwargs = {k: v for k, v in kwargs.items() if k in allowed_keys}
        return PostmanConfig(**filtered_kwargs)

    def get(self, key: str, default: Any = None) -> Any:
        """
        Get configuration value using dot‑notation.

        Example:
            manager.get("port") -> 8080
            manager.get("retry_policy.max_retries") -> 3
        """
        if self._config is None:
            self.load()
        # Split by dots
        parts = key.split(".")
        value = self._config
        for part in parts:
            if not hasattr(value, part):
                return default
            value = getattr(value, part)
        return value

    @traces(spec="002", story="FR-011", reqs=["REQ-002-FR11-005"])
    def validate(self, config: dict[str, Any]) -> tuple[bool, list[str]]:
        """
        Validate configuration dictionary.

        Returns:
            (is_valid, list_of_error_messages)
        """
        errors = []
        # Validate port range
        port = config.get("port")
        if port is not None and not (1 <= port <= 65535):
            errors.append(f"Port {port} out of valid range 1‑65535")
        # Validate max_message_size > 0
        max_msg = config.get("max_message_size")
        if max_msg is not None and max_msg <= 0:
            errors.append("max_message_size must be positive")
        # Validate max_queue_size > 0
        max_q = config.get("max_queue_size")
        if max_q is not None and max_q <= 0:
            errors.append("max_queue_size must be positive")
        # Validate default_ttl > 0
        ttl = config.get("default_ttl")
        if ttl is not None and ttl <= 0:
            errors.append("default_ttl must be positive")
        # Validate thread_pool_size > 0
        threads = config.get("thread_pool_size")
        if threads is not None and threads <= 0:
            errors.append("thread_pool_size must be positive")
        # Validate retry_policy structure if present
        retry = config.get("retry_policy")
        if retry is not None and not isinstance(retry, dict):
            errors.append("retry_policy must be a dictionary")
        return len(errors) == 0, errors

    @traces(spec="002", story="US-009", reqs=["REQ-002-US-009-028"])
    def on_change(self, callback: Callable[[str, Any, Any], None]) -> None:
        """Register a callback for configuration changes."""
        with self._lock:
            self._callbacks.append(callback)

    @traces(spec="002", story="US-009", reqs=["REQ-002-US-009-028"])
    def _notify_callbacks(
        self, old_config: PostmanConfig, new_config: PostmanConfig
    ) -> None:
        """Notify all registered callbacks about changed values."""
        for field in dataclasses.fields(PostmanConfig):
            old_val = getattr(old_config, field.name)
            new_val = getattr(new_config, field.name)
            if old_val != new_val:
                for cb in self._callbacks:
                    try:
                        cb(field.name, old_val, new_val)
                    except Exception as e:
                        logger.error(
                            f"Config callback failed for field '{field.name}': {e}"
                        )

    @traces(spec="002", story="FR-011", reqs=["REQ-002-FR11-011"])
    def _start_watch(self) -> None:
        """Start background thread that watches for config file changes."""
        if self._config_path is None:
            return
        self._stop_watch.clear()
        self._watch_thread = threading.Thread(
            target=self._watch_loop, daemon=True, name="ConfigWatcher"
        )
        self._watch_thread.start()

    @traces(spec="002", story="FR-011", reqs=["REQ-002-FR11-012"])
    def _watch_loop(self) -> None:
        """Loop that periodically checks the config file modification time."""
        while not self._stop_wait(self._watch_interval_seconds):
            if self._config_path is None:
                continue
            try:
                current_mtime = self._config_path.stat().st_mtime
                if current_mtime > self._last_mtime:
                    # File changed, trigger reload
                    self.reload()
                    self._last_mtime = current_mtime
            except (OSError, AttributeError) as e:
                logger.debug(f"Config file watch error (file may not exist): {e}")

    @traces(spec="002", story="FR-011", reqs=["REQ-002-FR11-012"])
    def _stop_wait(self, timeout: float) -> bool:
        """Wait for stop event with timeout. Returns True if stop requested."""
        return self._stop_watch.wait(timeout)

    @traces(spec="002", story="FR-011", reqs=["REQ-002-FR11-011"])
    def stop_watch(self) -> None:
        """Stop the background watch thread."""
        self._stop_watch.set()
        if self._watch_thread:
            self._watch_thread.join(
                timeout=float(CONFIG_WATCH_THREAD_JOIN_TIMEOUT_SECONDS)
            )