Source code for structum_lab.plugins.dynaconf.features.health

# src/structum_lab.plugins.dynaconf/health.py
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: 2025 PythonWoods

"""Configuration Health Check System.

Provides architectural components to perform runtime validation of the
configuration system integrity.
"""

import json
import logging
from abc import abstractmethod
from dataclasses import dataclass, field
from enum import Enum
from typing import TYPE_CHECKING, Any, Protocol, cast

if TYPE_CHECKING:
    from structum_lab.plugins.dynaconf.core.manager import ConfigManager

log = logging.getLogger(__name__)



[docs]
class HealthStatus(str, Enum):
    """Status levels for health checks."""

    HEALTHY = "healthy"
    DEGRADED = "degraded"
    UNHEALTHY = "unhealthy"




[docs]
@dataclass
class HealthCheckResult:
    """Result of a health check operation.

    Attributes:
        name: Identifier for the check.
        status: Overall health status.
        message: Human-readable status description.
        metadata: Additional diagnostic information.
    """

    name: str
    status: HealthStatus
    message: str
    metadata: dict[str, Any] = field(default_factory=dict)




[docs]
class HealthCheck(Protocol):
    """Protocol for a configuration health check component."""


[docs]
    @abstractmethod
    def check(self) -> HealthCheckResult:
        """Executes the check and returns the result."""
        pass





[docs]
class ConfigFileIntegrityCheck:
    """Verifies that persistence files are valid JSON if they exist."""


[docs]
    def __init__(self, config_manager: "ConfigManager") -> None:
        """Initialize the integrity check.

        Args:
            config_manager: The configuration manager containing registered builders.
        """
        self.manager = config_manager



[docs]
    def check(self) -> HealthCheckResult:
        """Execute the integrity check on all registered configurations.

        Returns:
            HealthCheckResult: The result of the integrity check.
        """
        try:
            # We access the internal builders map to check all registered configs
            # This logic assumes we are inside the package and allowed to access internals
            for config_name, builder_cls in self.manager._builders.items():
                builder = cast(Any, builder_cls)()

                # Check persistence file
                persistence_file = builder.get_persistence_file()
                if persistence_file.exists():
                    try:
                        content = persistence_file.read_text(encoding="utf-8")
                        json.loads(content)
                    except json.JSONDecodeError as e:
                        return HealthCheckResult(
                            name="config_file_integrity",
                            status=HealthStatus.UNHEALTHY,
                            message=f"Corrupted persistence file for '{config_name}' at {persistence_file}",
                            metadata={"error": str(e), "file": str(persistence_file)},
                        )
                    except Exception as e:
                        return HealthCheckResult(
                            name="config_file_integrity",
                            status=HealthStatus.UNHEALTHY,
                            message=f"Error reading persistence file for '{config_name}': {e}",
                            metadata={"file": str(persistence_file)},
                        )

            return HealthCheckResult(
                name="config_file_integrity",
                status=HealthStatus.HEALTHY,
                message="All configuration files are valid",
            )

        except Exception as e:
            return HealthCheckResult(
                name="config_file_integrity",
                status=HealthStatus.UNHEALTHY,
                message=f"System error during integrity check: {e}",
            )





[docs]
class PydanticValidationCheck:
    """Re-validates all loaded configuration models against their schemas."""


[docs]
    def __init__(self, config_manager: "ConfigManager") -> None:
        """Initialize with a reference to the config manager.

        Args:
            config_manager: The ConfigManager to check.
        """
        self.manager = config_manager



[docs]
    def check(self) -> HealthCheckResult:
        """Re-validate all loaded Pydantic models."""
        try:
            for config_name in self.manager._builders.keys():
                # get_config() retrieves the cached model instance
                # We force re-validation by dumping and re-validating
                try:
                    model = self.manager.get_config(config_name)
                    # Support Pydantic V2 and V1
                    if hasattr(model, "model_dump"):
                        data = model.model_dump()
                        model.model_validate(data)
                    elif hasattr(model, "dict[str, Any]"):
                        data = model.dict[str, Any]()
                        model.parse_obj(data)

                except Exception as e:
                    return HealthCheckResult(
                        name="pydantic_validation",
                        status=HealthStatus.UNHEALTHY,
                        message=f"Model validation failed for '{config_name}'",
                        metadata={"error": str(e), "config": config_name},
                    )

            return HealthCheckResult(
                name="pydantic_validation",
                status=HealthStatus.HEALTHY,
                message="All models pass validation",
            )

        except Exception as e:
            return HealthCheckResult(
                name="pydantic_validation",
                status=HealthStatus.UNHEALTHY,
                message=f"System error during validation check: {e}",
            )





[docs]
class HealthCheckRegistry:
    """Registry to manage and run multiple health checks."""


[docs]
    def __init__(self) -> None:
        """Initialize an empty health check registry."""
        self._checks: list[HealthCheck] = []



[docs]
    def register(self, check: HealthCheck) -> None:
        """Register a health check.

        Args:
            check: A health check instance to register.
        """
        self._checks.append(check)



[docs]
    def run_all(self) -> dict[str, HealthCheckResult]:
        """Executes all registered checks."""
        results = {}
        for check in self._checks:
            try:
                result = check.check()
                results[result.name] = result
            except Exception as e:
                # Catch-all for check execution failures
                log.exception(f"Health check failed unexpectedly: {check}")
                error_result = HealthCheckResult(
                    name=check.__class__.__name__,
                    status=HealthStatus.UNHEALTHY,
                    message=f"Check execution failed: {e}",
                )
                results[error_result.name] = error_result
        return results