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.
"""

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

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

log = logging.getLogger(__name__)



[docs]
class HealthStatus(str, Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    UNHEALTHY = "unhealthy"




[docs]
@dataclass
class HealthCheckResult:
    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"):
        self.manager = config_manager



[docs]
    def check(self) -> HealthCheckResult:
        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 = 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"):
        self.manager = config_manager



[docs]
    def check(self) -> HealthCheckResult:
        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"):
                        data = model.dict()
                        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):
        self._checks: List[HealthCheck] = []



[docs]
    def register(self, check: HealthCheck) -> None:
        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