from __future__ import annotations import os from dataclasses import dataclass, field from pathlib import Path from typing import Any import tomllib import yaml @dataclass class DiscoveryConfig: """Configuration for application discovery methods.""" file_based_enabled: bool = False file_based_path: Path = Path("/var/run/active-apps") http_enabled: bool = True http_host: str = "127.0.0.1" http_port: int = 8080 scan_enabled: bool = True scan_paths: list[Path] = field(default_factory=lambda: [Path("/home"), Path("/opt/apps")]) exclude_patterns: list[str] = field( default_factory=lambda: ["**/node_modules/**", "**/.venv/**", "**/venv/**"] ) @dataclass class DockerComposeConfig: """Configuration for Docker Compose service management.""" base_dir: Path = Path("/opt") compose_file: str = "docker-compose.yml" @dataclass class ServicesConfig: """Configuration for service lifecycle management.""" grace_period_minutes: int check_interval_seconds: int state_file: Path @dataclass class LoggingConfig: """Configuration for logging.""" level: str = "INFO" file: Path | None = None max_bytes: int = 10 * 1024 * 1024 backup_count: int = 5 @dataclass class ControllerConfig: """Main configuration for the infrastructure controller.""" discovery: DiscoveryConfig = field(default_factory=DiscoveryConfig) docker: DockerComposeConfig = field(default_factory=DockerComposeConfig) services: ServicesConfig = field( default_factory=lambda: ServicesConfig( grace_period_minutes=15, check_interval_seconds=60, state_file=Path("/var/lib/infra-controller/state.json"), ) ) logging: LoggingConfig = field(default_factory=LoggingConfig) @classmethod def from_file(cls, path: Path | str) -> ControllerConfig: """Load configuration from a TOML or YAML file.""" path = Path(path) if not path.exists(): return cls() data = load_config_file(path) return cls._from_dict(data) @classmethod def from_env(cls) -> ControllerConfig: """Load configuration from environment variables only.""" config = cls() apply_env_overrides(config) return config @classmethod def _from_dict(cls, data: dict[str, Any]) -> ControllerConfig: """Parse configuration from a dictionary.""" config = cls() parse_discovery_config(config, data.get("discovery")) parse_docker_config(config, data.get("docker")) parse_services_config(config, data.get("services")) parse_logging_config(config, data.get("logging")) return config def load_config_file(path: Path) -> dict[str, Any]: """Load configuration file based on extension.""" if path.suffix in {".toml", ".tml"}: with open(path, "rb") as f: return tomllib.load(f) if path.suffix in {".yml", ".yaml"}: with open(path, "r", encoding="utf-8") as f: return yaml.safe_load(f) or {} raise ValueError(f"Unsupported config format: {path.suffix}") def parse_discovery_config(config: ControllerConfig, data: Any) -> None: """Parse discovery configuration section.""" if not isinstance(data, dict): return # File-based discovery if isinstance(data.get("file_based"), dict): fb = data["file_based"] config.discovery.file_based_enabled = bool(fb.get("enabled", True)) if fb.get("path"): config.discovery.file_based_path = Path(fb["path"]) # HTTP discovery if isinstance(data.get("http"), dict): http = data["http"] config.discovery.http_enabled = bool(http.get("enabled", True)) config.discovery.http_host = str(http.get("host", config.discovery.http_host)) if http.get("port") is not None: config.discovery.http_port = int(http["port"]) # Scan-based discovery if isinstance(data.get("scan"), dict): scan = data["scan"] config.discovery.scan_enabled = bool(scan.get("enabled", True)) if isinstance(scan.get("paths"), list): config.discovery.scan_paths = [Path(p) for p in scan["paths"]] if isinstance(scan.get("exclude_patterns"), list): config.discovery.exclude_patterns = [str(p) for p in scan["exclude_patterns"]] def parse_docker_config(config: ControllerConfig, data: Any) -> None: """Parse Docker configuration section.""" if not isinstance(data, dict): return if data.get("base_dir"): config.docker.base_dir = Path(data["base_dir"]) if data.get("compose_file"): config.docker.compose_file = str(data["compose_file"]) def parse_services_config(config: ControllerConfig, data: Any) -> None: """Parse services configuration section.""" if not isinstance(data, dict): return if data.get("grace_period_minutes") is not None: config.services.grace_period_minutes = int(data["grace_period_minutes"]) if data.get("check_interval_seconds") is not None: config.services.check_interval_seconds = int(data["check_interval_seconds"]) if data.get("state_file"): config.services.state_file = Path(data["state_file"]) def parse_logging_config(config: ControllerConfig, data: Any) -> None: """Parse logging configuration section.""" if not isinstance(data, dict): return if data.get("level"): config.logging.level = str(data["level"]) if data.get("file"): config.logging.file = Path(data["file"]) if data.get("max_bytes") is not None: config.logging.max_bytes = int(data["max_bytes"]) if data.get("backup_count") is not None: config.logging.backup_count = int(data["backup_count"]) def apply_env_overrides(config: ControllerConfig) -> None: """Apply environment variable overrides to configuration.""" return def load_config(config_path: str | os.PathLike[str] | None = None) -> ControllerConfig: """ Load configuration with the following precedence: 1. Explicit config_path parameter 2. CONFIG_PATH environment variable 3. Default paths (/etc/infra-controller/config.{toml,yml}) 4. Default configuration Environment variables always override file-based configuration. """ # Determine config source if config_path is None: config_path = os.getenv("CONFIG_PATH") cfg = _load_base_config(config_path) # Apply environment variable overrides apply_env_overrides(cfg) return cfg def _load_base_config(config_path: str | os.PathLike[str] | None) -> ControllerConfig: """Load base configuration from file or defaults.""" if config_path: path = Path(str(config_path)) if path.exists(): return ControllerConfig.from_file(path) return ControllerConfig() # Try default locations for default_path in [ Path("/etc/infra-controller/config.toml"), Path("/etc/infra-controller/config.yml"), ]: if default_path.exists(): return ControllerConfig.from_file(default_path) # Fall back to defaults return ControllerConfig()