Configuration Management Patterns

---




name: configuration-management-patterns
description: Designs production configuration management with schema validation, hierarchical merging of config sources, hot reload capabilities, secret injection from vaults, and environment-specific defaults for reliable deployment.
license: MIT
compatibility: opencode
metadata:
  version: "1.0.0"
  domain: coding
  triggers: configuration management, config validation, schema validation, config hot reload, secret injection, HashiCorp Vault, Pydantic settings, how do i manage application configuration
  role: implementation
  scope: implementation
  output-format: code
  content-types: [code, guidance, do-dont, examples]
  related-skills: database-design-modeling, performance-optimization, security-review, cncf-deployment-orchestration
  author: https://github.com/openai/skill-router-contributors
  source: https://github.com/paulpas/git/agent-skill-router
  archetypes:
    - tactical
    - generation
  anti_triggers:
    - brainstorming
    - vague ideation
    - code golf
    - over-engineering
  response_profile:
    verbosity: low
    directive_strength: high
    abstraction_level: operational




---





# Configuration Management Patterns

Designs production-grade configuration management with schema validation, hierarchical merging of config sources, hot reload capabilities, secret injection from vaults, and environment-specific defaults for reliable deployment.

## TL;DR Checklist

- [ ] Define config schema with Pydantic Settings v2 using `model_config` and field-level validators
- [ ] Implement hierarchical merging: file defaults < env vars < secrets vault < runtime overrides
- [ ] Validate all required fields at startup — fail fast with a clear error listing missing keys
- [ ] Inject secrets from environment variables or vault — never hardcode in source or config files
- [ ] Support hot reload for non-secret config via filesystem watcher or KV store polling
- [ ] Write config fixture patterns for tests that isolate config from application logic

---

## When to Use

Use this skill when:

- Your application requires configuration that varies across environments (dev, staging, production)
- You need schema validation to catch misconfigurations at startup rather than runtime failures
- Secrets (API keys, database passwords) must be injected securely without appearing in source control
- Configuration changes should take effect without restarting the application (hot reload)
- Multiple config sources (files, env vars, vaults, CLI flags) need consistent hierarchical merging

---

## When NOT to Use

Avoid this skill for:

- Simple scripts or CLIs with fewer than 5 configuration values — use `argparse` or `click` defaults instead
- One-off batch jobs run in identical environments — hardcoded values are simpler and sufficient
- Configuration that changes per-request (feature flags served from a remote KV store) — use a dedicated feature flag service

---

## Core Workflow

1. **Define the Config Schema** — Create a Pydantic `BaseSettings` model with typed fields, defaults, and field-level descriptions. Use `model_config` for env prefix and secrets sources. **Checkpoint:** Every field must have an explicit type annotation — never use `Any` or omit the type hint.

2. **Set Up Hierarchical Merging** — Configure Pydantic Settings v2's `SecretsSettingsSource` and custom sources to merge from multiple origins. The order is: file defaults → environment variables → secrets vault → runtime overrides. **Checkpoint:** Verify that env vars override file values by testing with a conflicting value in the environment.

3. **Inject Secrets Securely** — Separate secret fields (passwords, API keys, tokens) from public config using `SecretStr`. Load them from environment variables or HashiCorp Vault at startup. **Checkpoint:** No secrets should appear in log output, error messages, or stack traces — mask all secret values with `****`.

4. **Implement Validation and Fail-Fast Startup** — Add model validators for cross-field constraints (e.g., `max_connections` > 0, TLS cert and key must both exist). Validate at application bootstrap. **Checkpoint:** Missing required config should cause an immediate startup failure with a list of exactly which keys are missing.

5. **Add Hot Reload for Mutable Config** — For non-secret config (feature flags, rate limits), add a filesystem watcher or KV polling mechanism that updates the config in-place. Secret config requires restart on change. **Checkpoint:** Hot reload must be thread-safe — use `atomic` replacement of the entire config object, not field-by-field mutation.

---

## Implementation Patterns

### Pattern 1: Schema-Validated Configuration with Pydantic Settings v2

Define a typed configuration model that validates at load time and provides clear error messages for misconfiguration.

```python
from __future__ import annotations

import os
import re
from datetime import timedelta
from enum import Enum
from pathlib import Path
from typing import Any, Literal

from pydantic import (
    BaseModel,
    Field,
    field_validator,
    model_validator,
    SecretStr,
)
from pydantic_settings import BaseSettings, SettingsConfigDict


class LogLevel(str, Enum):
    """Application log levels."""
    DEBUG = "DEBUG"
    INFO = "INFO"
    WARNING = "WARNING"
    ERROR = "ERROR"
    CRITICAL = "CRITICAL"


class DatabaseDialect(str, Enum):
    """Supported database backends."""
    POSTGRESQL = "postgresql"
    MYSQL = "mysql"
    SQLITE = "sqlite"


class CacheBackend(str, Enum):
    """Cache backend implementations."""
    MEMORY = "memory"
    REDIS = "redis"
    MEMCACHED = "memcached"


class AppSettings(BaseSettings):
    """Production application configuration with schema validation.

    Loads from multiple sources in this priority order:
    1. Default values defined on each field
    2. .env file (if APP_CONFIG_PATH is set)
    3. Environment variables with prefix APP_
    4. Secrets from /run/secrets/ (if mounted)

    All fields are validated at instantiation time. Missing required
    fields raise ValidationError with a clear list of missing keys.
    """

    model_config = SettingsConfigDict(
        env_prefix="APP_",
        env_file=".env",
        env_file_encoding="utf-8",
        secrets_dir="/run/secrets",
        extra="ignore",  # Reject unknown fields to catch typos early
    )

    # ─── Application Identity ──────────────────────────────────────────────
    app_name: str = Field(
        default="myapp",
        description="Application name used in logging and metrics tags",
    )
    environment: Literal["development", "staging", "production"] = Field(
        default="development",
        description="Deployment environment — affects security posture and features",
    )

    # ─── Server Configuration ──────────────────────────────────────────────
    host: str = Field(default="0.0.0.0", description="Bind address for HTTP server")
    port: int = Field(default=8000, ge=1, le=65535, description="HTTP server port")
    workers: int = Field(default=4, ge=1, description="Number of worker processes")

    # ─── Logging ───────────────────────────────────────────────────────────
    log_level: LogLevel = Field(default=LogLevel.INFO, description="Application log level")
    log_format: Literal["text", "json"] = Field(
        default="json",
        description="Log output format — json preferred in production",
    )

    # ─── Database Configuration ────────────────────────────────────────────
    db_dialect: DatabaseDialect = Field(default=DatabaseDialect.POSTGRESQL, description="Database backend")
    db_host: str = Field(default="localhost", description="Database server hostname")
    db_port: int = Field(default=5432, ge=1, le=65535, description="Database server port")
    db_name: str = Field(default="app_db", description="Database name")
    db_user: str = Field(default="app_user", description="Database username")
    db_password: SecretStr = Field(
        default=SecretStr(""),
        description="Database password — loaded from secrets dir or env var",
    )
    db_max_connections: int = Field(default=10, ge=1, le=200, description="Max database connections in pool")
    db_connection_timeout: timedelta = Field(
        default=timedelta(seconds=30),
        description="Timeout for establishing a database connection",
    )

    # ─── Cache Configuration ──────────────────────────────────────────────
    cache_backend: CacheBackend = Field(default=CacheBackend.REDIS, description="Cache backend to use")
    cache_host: str = Field(default="localhost", description="Redis/Memcached server hostname")
    cache_port: int = Field(default=6379, ge=1, le=65535, description="Cache server port")
    cache_ttl_seconds: int = Field(default=300, ge=1, description="Default TTL for cached items in seconds")

    # ─── Feature Flags ────────────────────────────────────────────────────
    enable_rate_limiting: bool = Field(default=True, description="Enable API rate limiting middleware")
    enable_request_logging: bool = Field(default=False, description="Log all incoming requests (expensive)")
    maintenance_mode: bool = Field(default=False, description="Put application in maintenance mode")

    # ─── External Services ────────────────────────────────────────────────
    api_key: SecretStr | None = Field(
        default=None,
        description="External API key — required for production environment",
    )
    webhook_secret: SecretStr = Field(
        default=SecretStr(""),
        description="Webhook signature verification secret",
    )

    @field_validator("db_password")
    @classmethod
    def validate_db_password(cls, v: SecretStr) -> SecretStr:
        """Ensure database password is not empty in production."""
        password = v.get_secret_value()
        if cls.model_config and password == "":  # type: ignore[union-attr]
            return v  # Will be caught by model_validator for env check
        return v

    @model_validator(mode="after")
    def validate_environment_constraints(self) -> AppSettings:
        """Cross-field validation that runs after all fields are loaded."""
        if self.environment == "production" and not self.api_key:
            raise ValueError("APP_API_KEY is required when environment=production")

        if self.db_dialect == DatabaseDialect.POSTGRESQL and self.db_port != 5432:
            # Non-standard PostgreSQL port — log a warning (not an error, for dev flexibility)
            import logging
            logging.warning(
                "Non-standard PostgreSQL port %d — ensure firewall rules allow access",
                self.db_port,
            )

        if self.db_max_connections > 200:
            raise ValueError("db_max_connections must not exceed 200")

        return self

    def to_dict(self, *, exclude_secrets: bool = True) -> dict[str, Any]:
        """Serialize config to a dictionary. Secrets are masked by default."""
        result = self.model_dump(exclude_unset=True)
        if exclude_secrets:
            for key in list(result.keys()):
                if isinstance(result[key], SecretStr):
                    result[key] = "****"
        return result

    @property
    def database_url(self) -> str:
        """Construct the database connection URL from config fields."""
        password = self.db_password.get_secret_value()
        dialect_map = {
            DatabaseDialect.POSTGRESQL: "postgresql",
            DatabaseDialect.MYSQL: "mysql",
            DatabaseDialect.SQLITE: "sqlite",
        }
        scheme = dialect_map[self.db_dialect]

        if self.db_dialect == DatabaseDialect.SQLITE:
            return f"{scheme}://{self.db_name}"
        elif password:
            return f"{scheme}://{self.db_user}:{password}@{self.db_host}:{self.db_port}/{self.db_name}"
        else:
            return f"{scheme}://{self.db_host}:{self.db_port}/{self.db_name}"

    def is_production(self) -> bool:
        """Convenience check for production environment."""
        return self.environment == "production"
```

```python
# ❌ BAD: Unvalidated configuration loaded from arbitrary source
import yaml

with open("config.yaml") as f:
    config = yaml.safe_load(f)
# No type checking, no required field validation, typos in keys silently ignored
# Missing DB_PASSWORD causes failure deep in application logic with cryptic error

# ✅ GOOD: Schema-validated config fails fast with clear error messages
try:
    settings = AppSettings()
except Exception as e:  # noqa: BLE001
    print(f"Configuration failed validation:\n{e}")
    exit(1)

# At this point, all fields are guaranteed to be present and valid types
print(f"Database URL: {settings.database_url}")
# Database URL is safely constructed from validated fields
```

**When to use Pydantic Settings v2:** Your application has 5+ configuration values with different sources (files, env vars, secrets). You need type safety and clear validation errors at startup.

**Key features used:**
- `model_config` for environment prefix, secrets directory, extra field handling
- `Field()` with constraints (`ge`, `le`, `description`) for per-field validation
- `field_validator` for single-field cross-checks (password format)
- `model_validator` for multi-field cross-checks (production requires API key)

---

### Pattern 2: Hierarchical Config Merging

Merge configuration from multiple sources with a well-defined priority order. Lower-priority sources provide defaults; higher-priority sources override them.

```python
from __future__ import annotations

import json
import logging
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

logger = logging.getLogger(__name__)


@dataclass
class ConfigSource:
    """Represents a single configuration source with its priority level.

    Lower `priority` value = higher precedence (overrides lower-priority sources).
    Sources are merged top-down; later sources override earlier ones for the same key.
    """
    name: str
    priority: int
    data: dict[str, Any] = field(default_factory=dict)
    is_valid: bool = True
    error_message: str | None = None


class HierarchicalConfigMerger:
    """Merges configuration from multiple sources with deterministic priority ordering.

    Priority order (lowest to highest):
      1. Built-in defaults (hardcoded in code)
      2. Configuration file (.yaml, .json, .toml)
      3. Environment variables (prefixed, converted to dotted notation)
      4. Secrets vault values
      5. Runtime overrides (programmatic API calls)

    Nested dictionaries are merged recursively; scalar values are overwritten.
    """

    def __init__(self, strict_mode: bool = True) -> None:
        self._sources: list[ConfigSource] = []
        self._strict_mode = strict_mode
        self._merged: dict[str, Any] = {}

    @property
    def sources(self) -> list[ConfigSource]:
        return list(self._sources)

    def add_source(self, source: ConfigSource) -> "HierarchicalConfigMerger":
        """Add a configuration source. Sources are sorted by priority (ascending = highest precedence last)."""
        self._sources.append(source)
        self._sources.sort(key=lambda s: s.priority)
        return self

    def merge(self) -> dict[str, Any]:
        """Merge all sources in priority order. Returns the final merged configuration."""
        if not self._sources:
            self._merged = {}
            return self._merged

        self._merged = {}

        for source in self._sources:
            if not source.is_valid:
                logger.warning("Skipping invalid config source '%s': %s", source.name, source.error_message)
                continue

            self._merged = self._deep_merge(self._merged, source.data)
            logger.info("Applied source '%s' (priority=%d): %d keys merged", source.name, source.priority, len(source.data))

        return dict(self._merged)  # Return a copy to prevent mutation

    def _deep_merge(self, base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]:
        """Recursively merge two dictionaries. Scalar values in override replace base."""
        result = dict(base)  # Shallow copy of base

        for key, value in override.items():
            if (
                key in result
                and isinstance(result[key], dict)
                and isinstance(value, dict)
            ):
                # Recursively merge nested dictionaries
                result[key] = self._deep_merge(result[key], value)
            else:
                # Scalar or list values — override wins completely
                result[key] = value

        return result

    def validate_required_keys(self, required_keys: list[str]) -> list[str]:
        """Check that all required configuration keys are present after merging.

        Supports dotted notation for nested keys (e.g., "database.host").

        Args:
            required_keys: Flat list of required key paths using dot notation

        Returns:
            List of missing keys (empty if all present)
        """
        missing = []
        for key_path in required_keys:
            parts = key_path.split(".")
            current = self._merged

            for part in parts:
                if isinstance(current, dict) and part in current:
                    current = current[part]
                else:
                    missing.append(key_path)
                    break

        if missing:
            logger.error("Missing required configuration keys: %s", ", ".join(missing))
        return missing

    def get(self, key_path: str, default: Any = None) -> Any:
        """Get a value from the merged config using dotted key path notation.

        Args:
            key_path: Dotted path to the configuration value (e.g., "database.host")
            default: Default value if the key is not found

        Returns:
            The configuration value or the provided default
        """
        parts = key_path.split(".")
        current: Any = self._merged

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

        return current


def build_env_var_source(prefix: str = "APP_") -> dict[str, Any]:
    """Convert environment variables with a given prefix into a flat dictionary.

    Converts dotted notation in env var names to nested dictionaries:
      APP_DATABASE_HOST → {"database": {"host": "..."}}
      APP_LOG_LEVEL → {"log_level": "..."}
    """
    result: dict[str, Any] = {}
    for key, value in sorted(os.environ.items()):  # type: ignore[name-defined]
        if key.startswith(prefix):
            # Strip prefix and convert to dotted lowercase notation
            clean_key = key[len(prefix):].lower()
            parts = clean_key.split("_")

            # Insert into nested structure
            current = result
            for part in parts[:-1]:
                if part not in current or not isinstance(current[part], dict):
                    current[part] = {}
                current = current[part]
            current[parts[-1]] = value

    return result


def load_file_source(
    path: str | Path,
    format_hint: str = "yaml",
) -> ConfigSource:
    """Load configuration from a file (YAML or JSON)."""
    path = Path(path) if isinstance(path, str) else path

    if not path.exists():
        return ConfigSource(
            name=str(path),
            priority=10,
            is_valid=False,
            error_message=f"File not found: {path}",
        )

    try:
        content = path.read_text(encoding="utf-8")
        if format_hint == "json":
            data = json.loads(content)
        else:  # default to YAML
            import yaml
            data = yaml.safe_load(content) or {}

        return ConfigSource(name=str(path), priority=10, data=data)

    except Exception as e:  # noqa: BLE001
        return ConfigSource(
            name=str(path),
            priority=10,
            is_valid=False,
            error_message=f"Failed to parse {format_hint} file: {e}",
        )
```

```python
# ❌ BAD: Single source of truth — no hierarchy, no override mechanism
# All config in one file, no way to override per environment without editing the file
import yaml
config = yaml.safe_load(open("config.yaml"))  # No fallbacks, no env var overrides

# ✅ GOOD: Hierarchical merging with clear priority and validation
from pathlib import Path
from pydantic_settings import BaseSettings

merger = HierarchicalConfigMerger()

# Source 1: Built-in defaults (highest priority = lowest number)
defaults = ConfigSource(name="defaults", priority=0, data={
    "host": "0.0.0.0",
    "port": 8000,
    "workers": 4,
})

# Source 2: YAML config file
file_source = load_file_source(Path("/etc/myapp/config.yaml"), format_hint="yaml")

# Source 3: Environment variables
env_source = ConfigSource(
    name="env_vars",
    priority=100,
    data=build_env_var_source(prefix="APP_"),
)

merger.add_source(defaults).add_source(file_source).add_source(env_source)
merged_config = merger.merge()

# Validate required keys before using
missing = merger.validate_required_keys([
    "db_dialect", "db_host", "db_name",
    "cache_backend", "api_key",
])
if missing:
    raise RuntimeError(f"Missing config: {missing}")
```

**When to use Hierarchical Merging:** Configuration needs to vary across environments without duplicating entire config files. Sensitive values (secrets) are loaded separately from public configuration.

---

### Pattern 3: Hot Reload with Filesystem Watchers

Non-secret configuration should take effect without restarting the application. This pattern uses a filesystem watcher to detect config file changes and atomically replace the in-memory config object.

```python
from __future__ import annotations

import asyncio
import logging
import signal
import threading
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable

logger = logging.getLogger(__name__)


@dataclass
class ConfigChange:
    """Represents a detected configuration change event."""
    source_path: Path
    change_type: str  # "created", "modified", "deleted"
    timestamp: float = field(default_factory=time.monotonic)


class HotReloadConfigManager:
    """Manages hot reloading of non-secret configuration.

    Watches config files for changes and atomically replaces the
    in-memory configuration when detected. Thread-safe via asyncio event loop.

    Secret configuration is NOT hot-reloaded — it requires a process restart
    to ensure old secret references are fully garbage collected.
    """

    def __init__(
        self,
        config_path: str | Path,
        reload_interval_seconds: float = 2.0,
        debounce_ms: int = 500,
    ) -> None:
        self._config_path = Path(config_path) if isinstance(config_path, str) else config_path
        self._reload_interval = reload_interval_seconds
        self._debounce_ms = debounce_ms
        self._last_modified: float = 0.0
        self._config: dict[str, Any] = {}
        self._callbacks: list[Callable[[dict[str, Any]], None]] = []
        self._running = False
        self._task: asyncio.Task | None = None
        self._file_hash: str | None = None

    @property
    def config(self) -> dict[str, Any]:
        """Thread-safe read access to the current configuration snapshot."""
        return dict(self._config)  # Return a copy to prevent mutation

    @property
    def is_running(self) -> bool:
        return self._running

    async def start(self, initial_config: dict[str, Any] | None = None) -> None:
        """Start watching for configuration changes."""
        if self._running:
            logger.warning("Hot reload manager already running — stopping first")
            await self.stop()

        # Load initial config
        if initial_config is not None:
            self._config = initial_config
        elif self._config_path.exists():
            self._config = self._load_current_config()

        self._running = True
        self._task = asyncio.create_task(self._watch_loop())
        logger.info("Hot reload started watching: %s", self._config_path)

    async def stop(self) -> None:
        """Stop the hot reload watcher."""
        if self._task and not self._task.done():
            self._task.cancel()
            try:
                await self._task
            except asyncio.CancelledError:
                pass
        self._running = False
        logger.info("Hot reload stopped")

    def on_config_change(self, callback: Callable[[dict[str, Any]], None]) -> None:
        """Register a callback invoked when config changes are detected.

        The callback receives the new configuration dictionary.
        Callbacks run synchronously — use asyncio.create_task() for async callbacks.
        """
        self._callbacks.append(callback)

    def _compute_file_hash(self, path: Path) -> str | None:
        """Compute a hash of the config file for change detection."""
        if not path.exists():
            return None
        import hashlib
        content = path.read_bytes()
        return hashlib.sha256(content).hexdigest()[:16]

    def _load_current_config(self) -> dict[str, Any]:
        """Load the current config file contents."""
        try:
            content = self._config_path.read_text(encoding="utf-8")
            if str(self._config_path).endswith(".json"):
                import json
                return json.loads(content)
            else:
                import yaml
                return yaml.safe_load(content) or {}
        except Exception as e:  # noqa: BLE001
            logger.error("Failed to load config from %s: %s", self._config_path, e)
            return {}

    async def _watch_loop(self) -> None:
        """Main watch loop: poll for file changes with debounce."""
        while self._running:
            try:
                current_hash = self._compute_file_hash(self._config_path)

                if current_hash and current_hash != self._file_hash:
                    # File changed — apply debounce to avoid reloading mid-write
                    import time as _time
                    now = _time.monotonic()
                    if now - self._last_modified > self._debounce_ms / 1000.0:
                        await self._apply_reload(current_hash)

            except asyncio.CancelledError:
                break
            except Exception as e:  # noqa: BLE001
                logger.error("Error in watch loop: %s", e)

            await asyncio.sleep(self._reload_interval)

    async def _apply_reload(self, new_hash: str) -> None:
        """Apply a detected configuration change."""
        old_config = dict(self._config)
        new_config = self._load_current_config()

        if not new_config:
            logger.warning("Reload produced empty config — keeping previous config")
            return

        # Check for structural differences before notifying callbacks
        if old_config == new_config:
            return  # No actual change detected (e.g., whitespace-only modification)

        # Atomically replace the config snapshot
        self._config = new_config
        self._file_hash = new_hash
        self._last_modified = time.monotonic()

        logger.info(
            "Configuration reloaded from %s (%d keys)",
            self._config_path,
            len(new_config),
        )

        # Notify registered callbacks
        for callback in self._callbacks:
            try:
                callback(new_config)
            except Exception as e:  # noqa: BLE001
                logger.error("Config change callback failed: %s", e)

    @property
    def last_modified(self) -> float:
        return self._last_modified
```

```python
# ❌ BAD: No hot reload — every config change requires a full application restart
async def main():
    config = load_config()  # Loaded once at startup, never refreshed
    await run_server(config)  # Config changes on disk have zero effect

# ✅ GOOD: Hot reload with atomic swap and callback notification
async def main():
    config_manager = HotReloadConfigManager(
        config_path="/etc/myapp/runtime.yaml",
        debounce_ms=500,
    )

    @config_manager.on_config_change
    async def on_config_updated(new_config: dict[str, Any]) -> None:
        # Update rate limiter, feature flags, or other dynamic behavior
        await update_rate_limiters(new_config.get("rate_limits", {}))
        await toggle_feature_flags(new_config.get("feature_flags", {}))

    await config_manager.start(initial_config={"rate_limits": {}, "feature_flags": {}})
    try:
        await run_server(config_manager.config)  # Reads current snapshot on each request
    finally:
        await config_manager.stop()
```

**When to use Hot Reload:** Non-secret configuration changes frequently (feature flags, rate limits, UI settings). Zero-downtime updates are required. Changes should propagate within seconds.

**When NOT to use Hot Reload:** Secret configuration — old secret references may persist in memory and connection pools. Use full restart for secret changes. The application has no mechanism to gracefully handle config-driven state transitions (e.g., changing port number requires socket rebind).

---

### Pattern 4: Secret Injection from Environment vs Vault

Separate secret management from public configuration. Load secrets from the most secure source available — environment variables, mounted files, or HashiCorp Vault — with automatic fallback.

```python
from __future__ import annotations

import json
import logging
import os
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

logger = logging.getLogger(__name__)


class SecretProvider(ABC):
    """Protocol for secret providers that fetch secrets from external sources."""

    @abstractmethod
    async def get_secret(self, key: str) -> str | None: ...

    @abstractmethod
    async def get_secrets_batch(self, keys: list[str]) -> dict[str, str]: ...

    @property
    @abstractmethod
    def name(self) -> str: ...


class EnvironmentSecretProvider(SecretProvider):
    """Loads secrets from environment variables.

    Keys are mapped by converting snake_case to UPPER_CASE with a given prefix.
    e.g., key="database_password" → env var "APP_DATABASE_PASSWORD"
    """

    def __init__(self, prefix: str = "APP_") -> None:
        self._prefix = prefix

    @property
    def name(self) -> str:
        return f"env({self._prefix})"

    async def get_secret(self, key: str) -> str | None:
        env_key = f"{self._prefix}{key.upper()}"
        value = os.environ.get(env_key)
        if value:
            logger.debug("Loaded secret from env var for key: %s", key[:8] + "****")
        return value

    async def get_secrets_batch(self, keys: list[str]) -> dict[str, str]:
        return {key: await self.get_secret(key) or "" for key in keys}


class FileSecretProvider(SecretProvider):
    """Loads secrets from individually-named files (mounted by Kubernetes or Docker).

    Each secret is stored as a file containing the raw secret value.
    Common pattern: /run/secrets/database_password, /run/secrets/api_key
    """

    def __init__(self, secrets_dir: str = "/run/secrets") -> None:
        self._secrets_dir = Path(secrets_dir)

    @property
    def name(self) -> str:
        return f"file({self._secrets_dir})"

    async def get_secret(self, key: str) -> str | None:
        secret_path = self._secrets_dir / key
        if not secret_path.exists():
            return None
        try:
            value = secret_path.read_text(encoding="utf-8").strip()
            logger.debug("Loaded secret from file for key: %s", key[:8] + "****")
            return value
        except Exception as e:  # noqa: BLE001
            logger.error("Failed to read secret file %s: %s", secret_path, e)
            return None

    async def get_secrets_batch(self, keys: list[str]) -> dict[str, str]:
        secrets = {}
        for key in keys:
            value = await self.get_secret(key)
            if value is not None:
                secrets[key] = value
        return secrets


class VaultSecretProvider(SecretProvider):
    """Loads secrets from HashiCorp Vault using the Kubernetes auth method or token auth.

    Supports both kv-v1 and kv-v2 secret engines with versioned retrieval.
    In production, use Kubernetes service account tokens for automatic authentication.
    """

    def __init__(
        self,
        vault_address: str = "https://vault.example.com:8200",
        vault_path: str = "secret",
        vault_role: str | None = None,
        vault_token: str | None = None,
    ) -> None:
        self._address = vault_address.rstrip("/")
        self._path = vault_path
        self._role = vault_role
        self._token = vault_token
        self._client: Any = None  # type: ignore[assignment]

    @property
    def name(self) -> str:
        return f"vault({self._address}/{self._path})"

    async def _get_client(self) -> Any:  # noqa: ANN401
        """Lazy-initialize the Vault client. Returns None if auth fails."""
        if self._client is not None:
            return self._client

        try:
            import hvac  # type: ignore[import-not-found]
            if self._token:
                self._client = hvac.Client(url=self._address, token=self._token)
            elif self._role:
                # Kubernetes auth — assumes running inside a pod with service account
                k8s_token = Path("/var/run/secrets/kubernetes.io/serviceaccount/token").read_text()  # noqa: PTH118
                jwt_data = {"jwt": k8s_token, "role": self._role}
                auth_resp = hvac.Client(url=self._address).auth.kubernetes.login(**jwt_data)
                self._client = hvac.Client(
                    url=self._address,
                    token=auth_resp["auth"]["client_token"],
                )
            else:
                logger.warning("Vault provider configured without token or role — skipping")
                return None
        except ImportError:
            logger.warning("hvac package not installed — Vault integration disabled")
            return None

        return self._client

    async def get_secret(self, key: str) -> str | None:
        """Retrieve a single secret from Vault."""
        client = await self._get_client()
        if client is None:
            return None

        try:
            # kv-v2 engine — include version=-1 for latest version
            path = f"{self._path}/{key}"
            resp = client.secrets.kv.v2.read_secret_version(path=path)  # type: ignore[attr-defined]
            data = resp["data"]["data"]
            return str(data.get("value") or data.get(key, ""))
        except Exception as e:  # noqa: BLE001
            logger.debug("Vault secret not found or error for key '%s': %s", key, e)
            return None

    async def get_secrets_batch(self, keys: list[str]) -> dict[str, str]:
        """Retrieve multiple secrets in one Vault API call."""
        client = await self._get_client()
        if client is None:
            return {}

        try:
            # kv-v1 supports bulk read with a single path prefix
            secrets_data = client.secrets.kv.v1.read_secret(path=self._path, all=True)  # type: ignore[attr-defined]
            result = {}
            for key in keys:
                if key in secrets_data.get("data", {}):  # type: ignore[index]
                    result[key] = str(secrets_data["data"][key])  # type: ignore[index]
            return result
        except Exception as e:  # noqa: BLE001
            logger.warning("Vault batch retrieval failed — falling back to individual lookups: %s", e)
            secrets = {}
            for key in keys:
                val = await self.get_secret(key)
                if val is not None:
                    secrets[key] = val
            return secrets


class SecretInjector:
    """Orchestrates secret loading across multiple providers with fallback.

    Tries each provider in order until the secret is found. Logs which provider
    supplied each secret for debugging (without exposing the actual secret value).
    """

    def __init__(self, providers: list[SecretProvider] | None = None) -> None:
        self._providers = providers or [
            VaultSecretProvider(),
            FileSecretProvider(secrets_dir="/run/secrets"),
            EnvironmentSecretProvider(prefix="APP_"),
        ]
        # Sort by priority: Vault > Files > Env vars (most secure first)
        provider_order = {"vault": 0, "file": 1, "env": 2}
        self._providers.sort(key=lambda p: provider_order.get(p.name.split("(")[0], 99))

    async def inject_secrets(self, secret_keys: list[str]) -> dict[str, str]:
        """Load multiple secrets from the available providers.

        Each key is fetched from the first provider that has it.
        Returns a dictionary of key → value for all successfully loaded secrets.

        Args:
            secret_keys: List of secret names to inject (e.g., ["database_password", "api_key"])

        Returns:
            Dictionary mapping secret keys to their values. Missing keys are omitted.
        """
        result: dict[str, str] = {}
        missing: list[str] = []

        for key in secret_keys:
            if key in result:
                continue  # Already loaded from a higher-priority provider

            found = False
            for provider in self._providers:
                value = await provider.get_secret(key)
                if value is not None:
                    result[key] = value
                    logger.info("Secret '%s' loaded via %s", key, provider.name)
                    found = True
                    break

            if not found:
                missing.append(key)

        if missing:
            logger.error(
                "Failed to load secrets from any provider: %s. "
                "Ensure they exist in env vars, /run/secrets/, or Vault.",
                ", ".join(missing),
            )

        return result

    @property
    def provider_names(self) -> list[str]:
        return [p.name for p in self._providers]
```

```python
# ❌ BAD: Secrets hardcoded in source code or config files tracked in git
DATABASE_PASSWORD = "super_secret_123"  # Exposed in version control!
API_KEY = os.getenv("API_KEY", "dev_key_for_local")  # Weak default for production risk

# ✅ GOOD: Secret injection with multi-source fallback and masking in logs
async def bootstrap_app():
    injector = SecretInjector([
        VaultSecretProvider(vault_address="https://vault.prod.internal:8200"),
        FileSecretProvider(secrets_dir="/run/secrets"),
        EnvironmentSecretProvider(prefix="APP_"),
    ])

    secrets = await injector.inject_secrets([
        "database_password",
        "api_key",
        "webhook_secret",
        "jwt_signing_key",
    ])

    # Use secrets in config — they are masked in logs automatically
    settings = AppSettings(
        db_password=SecretStr(secrets.get("database_password", "")),
        api_key=SecretStr(secrets.get("api_key")),
        webhook_secret=SecretStr(secrets.get("webhook_secret")),
    )
    # jwt_signing_key can be used directly for token validation
```

**When to use Vault:** Production environments with centralized secret management. Teams need audit trails for secret access, automated rotation, and RBAC on secret paths.

**When to use Environment Variables:** Simpler deployments (containers, serverless). Secrets are injected by the deployment platform (Kubernetes secrets as env vars, AWS Lambda environment variables).

**When to use Mounted Files:** Kubernetes-native pattern where each secret is a file in `/run/secrets/`. Cleaner than env vars for multi-line secrets and avoids shell escaping issues.

---

### Pattern 5: Configuration Testing with Fixture Patterns

Test your application's configuration loading and validation in isolation. Use fixtures that provide known-good config for unit tests and known-bad config for failure case tests.

```python
from __future__ import annotations

import os
import tempfile
from pathlib import Path
from typing import Any

import pytest
from pydantic import SecretStr


@pytest.fixture
def valid_config_dict() -> dict[str, Any]:
    """Provide a minimal valid configuration dictionary for testing."""
    return {
        "app_name": "test-app",
        "environment": "staging",
        "host": "127.0.0.1",
        "port": 8765,
        "workers": 2,
        "log_level": "WARNING",
        "log_format": "text",
        "db_dialect": "sqlite",
        "db_name": ":memory:",
        "db_max_connections": 5,
        "cache_backend": "memory",
        "cache_ttl_seconds": 60,
        "enable_rate_limiting": True,
    }


@pytest.fixture
def production_config_dict(valid_config_dict: dict[str, Any]) -> dict[str, Any]:
    """Provide a production-valid configuration with all required secrets."""
    config = {**valid_config_dict}
    config["environment"] = "production"
    config["api_key"] = "test-api-key-for-qa-only"
    config["workers"] = 8
    return config


@pytest.fixture
def invalid_missing_api_key_config(valid_config_dict: dict[str, Any]) -> dict[str, Any]:
    """Configuration that fails validation because it's missing production-required api_key."""
    config = {**valid_config_dict}
    config["environment"] = "production"
    # Intentionally omit api_key — should trigger ValidationError
    return config


@pytest.fixture
def invalid_negative_port_config(valid_config_dict: dict[str, Any]) -> dict[str, Any]:
    """Configuration that fails validation because port is out of range."""
    config = {**valid_config_dict}
    config["port"] = -1  # Below ge=1 constraint
    return config


@pytest.fixture
def invalid_max_connections_config(valid_config_dict: dict[str, Any]) -> dict[str, Any]:
    """Configuration that fails the model_validator for exceeding max connections."""
    config = {**valid_config_dict}
    config["db_max_connections"] = 500  # Above le=200 constraint and model_validator check
    return config


@pytest.fixture
def env_config_snapshot(monkeypatch: Any, valid_config_dict: dict[str, Any]) -> None:  # type: ignore[unused-ignores]
    """Set environment variables for a staging configuration during tests.

    Cleans up all APP_ prefixed environment variables after the test.
    """
    for key, value in valid_config_dict.items():
        monkeypatch.setenv(f"APP_{key.upper()}", str(value))


@pytest.fixture
def config_file(tmp_path: Path, valid_config_dict: dict[str, Any]) -> Path:  # type: ignore[unused-ignores]
    """Create a temporary YAML config file for testing file-based config loading."""
    import yaml

    config_path = tmp_path / "test_config.yaml"
    config_data = {**valid_config_dict}
    # Remove __pycache__ and other non-dict values that yaml can't serialize
    config_data.pop("api_key", None)

    config_path.write_text(yaml.dump(config_data), encoding="utf-8")
    return config_path


class TestAppSettings:
    """Test suite for AppSettings configuration validation.

    Each test validates a specific constraint or error path.
    """

    def test_default_config_loads_successfully(self, valid_config_dict: dict[str, Any]) -> None:
        """Valid configuration with SQLite should load without errors."""
        from pydantic_settings import SettingsConfigDict
        from app.settings import AppSettings

        settings = AppSettings(**valid_config_dict)
        assert settings.environment == "staging"
        assert settings.db_dialect.value == "sqlite"
        assert settings.is_production() is False

    def test_database_url_construction(self, valid_config_dict: dict[str, Any]) -> None:
        """Database URL must be correctly constructed from config fields."""
        from app.settings import AppSettings

        settings = AppSettings(**valid_config_dict)
        url = settings.database_url
        assert "sqlite" in url
        assert ":" in url  # :memory: SQLite special connection string
        assert "****" not in url  # No secrets leaked into the URL representation for SQLite

    def test_production_requires_api_key(self, invalid_missing_api_key_config: dict[str, Any]) -> None:
        """Production environment must have api_key — validation fails without it."""
        from pydantic import ValidationError
        from app.settings import AppSettings

        with pytest.raises(ValidationError) as exc_info:
            AppSettings(**invalid_missing_api_key_config)

        error_messages = [e["msg"] for e in exc_info.value.errors()]
        assert any("api_key" in msg.lower() for msg in error_messages)

    def test_port_range_validation(self, invalid_negative_port_config: dict[str, Any]) -> None:
        """Port must be between 1 and 65535 — validation rejects out-of-range values."""
        from pydantic import ValidationError
        from app.settings import AppSettings

        with pytest.raises(ValidationError) as exc_info:
            AppSettings(**invalid_negative_port_config)

        assert exc_info.value.error_count() == 1
        error = exc_info.value.errors()[0]
        assert error["loc"] == ("port",)

    def test_secrets_are_masked_in_output(self, valid_config_dict: dict[str, Any]) -> None:
        """SecretStr fields must be masked when serialized to dictionary."""
        from app.settings import AppSettings

        settings = AppSettings(**valid_config_dict)
        config_dict = settings.to_dict(exclude_secrets=True)
        assert "****" in str(config_dict.get("db_password", ""))
        # Public fields are NOT masked
        assert config_dict["app_name"] == "test-app"


class TestHotReloadConfigManager:
    """Test suite for configuration hot reload."""

    def test_config_reload_on_file_change(
        self,
        tmp_path: Path,
        config_file: Path,
    ) -> None:  # type: ignore[unused-ignores]
        """When config file changes on disk, the manager detects and applies the new config."""
        import yaml

        async def run_reload_test() -> None:
            from app.config_hotreload import HotReloadConfigManager

            loaded_configs: list[dict[str, Any]] = []
            manager = HotReloadConfigManager(
                config_path=config_file,
                reload_interval_seconds=0.1,
                debounce_ms=50,
            )

            @manager.on_config_change
            def on_change(new_config: dict[str, Any]) -> None:  # noqa: ANN202
                loaded_configs.append(dict(new_config))

            await manager.start(initial_config={"key": "initial"})
            assert manager.config["key"] == "initial"

            # Simulate file change
            new_data = {"key": "updated", "new_key": "value"}
            config_file.write_text(yaml.dump(new_data), encoding="utf-8")

            # Wait for debounce + reload cycle
            await asyncio.sleep(0.3)  # noqa: ASYNC251

            assert manager.config["key"] == "updated"
            assert len(loaded_configs) >= 1
            assert loaded_configs[0]["new_key"] == "value"

            await manager.stop()

        import asyncio
        asyncio.run(run_reload_test())
```

```python
# ❌ BAD: No configuration fixtures — tests use real config with live database connections
def test_my_feature():
    settings = AppSettings()  # Reads real production config → hits real database
    result = my_feature(settings)
    assert result is True  # Fragile, slow, non-reproducible

# ✅ GOOD: Configuration fixtures provide controlled, isolated config for tests
def test_my_feature(valid_config_dict):  # pytest fixture with SQLite in-memory DB
    settings = AppSettings(**valid_config_dict)
    result = my_feature(settings)
    assert result is True  # Fast, reproducible, no external dependencies
```

**When to use Config Fixtures:** Every integration test that touches configuration-dependent code. Unit tests for config validation logic specifically need both valid and invalid fixture variants.

**Fixture categories:**
- `valid_config_dict`: Minimal working config with SQLite in-memory backend
- `production_config_dict`: Full production config including required secrets
- `invalid_*_config`: Variants designed to trigger specific validation failures
- `env_config_snapshot`: Sets environment variables for testing env-var-based config loading
- `config_file`: Creates temporary file-based configs for hot reload testing

---

## Constraints

### MUST DO
- Use Pydantic Settings v2 with `model_config` — never use the deprecated `SettingsConfigDict` in class-level `__init__` or raw dict parsing
- Define all field types explicitly — no `Any`, no implicit type coercion from strings without validators
- Mask secret values in all log output, error messages, and serialized dictionaries (`to_dict(exclude_secrets=True)`)
- Validate cross-field constraints with `model_validator(mode="after")` — not with individual `field_validator` calls
- Use environment variable prefix consistently (e.g., `APP_`) to avoid collisions between applications
- Write pytest fixtures for both valid and invalid configurations

### MUST NOT DO
- Never store secret values in plain-text config files tracked in version control
- Do not use `os.environ.get()` directly in business logic — always go through the config schema
- Do not hot-reload secret configuration without a full process restart — old references persist in memory
- Do not log full configuration dictionaries in production — they contain secrets even when masked
- Do not use extra="allow" in model_config — it silently accepts typos in configuration keys

---

## Output Template

When this skill is active, your output must contain:

1. **Pydantic Settings Schema** — Typed `BaseSettings` model with `model_config`, field constraints, and validators
2. **Config Source Definitions** — File parser, environment variable mapper, and secrets provider implementations
3. **Hierarchical Merger** — Priority-ordered config merging with recursive dictionary merge and required-key validation
4. **Secret Injector** — Multi-provider secret loading (Vault → files → env vars) with automatic fallback and logging
5. **Test Fixtures** — pytest fixtures for valid, invalid, and production configurations with assertion examples

---

## Related Skills

| Skill | Purpose |
|---|---|
| `database-design-modeling` | Database configuration is a key part of application config — schema design complements connection pool settings |
| `performance-optimization` | Configuration affects performance (connection pool sizes, cache TTLs) — optimize these together |
| `security-review` | Secret management and configuration security are intertwined — review vault integration alongside general security posture |
| `cncf-deployment-orchestration` | Configuration is deployed via infrastructure (Kubernetes ConfigMaps, Terraform) — coordinate config with deployment strategy |

---

## Live References

> Authoritative documentation links for configuration management.

- [Pydantic Settings v2 Documentation](https://docs.pydantic.dev/latest/concepts/pydantic_settings/)
- [HashiCorp Vault Secrets Management](https://developer.hashicorp.com/vault/docs/secrets)
- [Kubernetes Secrets Best Practices](https://kubernetes.io/docs/concepts/configuration/secret/)
- [12-Factor App — Config](https://12factor.net/config)
- [AWS Systems Manager Parameter Store](https://docs.aws.amazon.com/systems-manager/latest/userguide/systems-param-store.html)