7eaea39928
Some checks failed
Integration Tests / integration-tests (push) Failing after 20s
Integration Tests / performance-tests (push) Has been skipped
Service Adapters (Python FastAPI) / test (3.11) (push) Failing after 24s
Service Adapters (Python FastAPI) / test (3.12) (push) Failing after 25s
Service Adapters (Python FastAPI) / test (3.13) (push) Failing after 25s
Service Adapters (Python FastAPI) / build (push) Has been skipped
### Summary of Changes - Removed unnecessary whitespace and standardized formatting in multiple files, including `main.py`, `logging_middleware.py`, `general.py`, and various health checker implementations. - Enhanced readability and maintainability of the codebase by ensuring consistent formatting practices. ### Expected Results - Improved code clarity, making it easier for developers to read and understand the service adapters' code. - Streamlined the codebase, facilitating future updates and maintenance.
127 lines
3.9 KiB
Python
127 lines
3.9 KiB
Python
"""
|
|
Health Checker Registry
|
|
|
|
This module provides a registry and factory for different health checker types.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Any, Dict, Optional, Type
|
|
|
|
from .api_checker import APIHealthChecker
|
|
from .base import BaseHealthChecker
|
|
from .custom_checker import CustomHealthChecker
|
|
from .sensor_checker import SensorHealthChecker
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class HealthCheckerRegistry:
|
|
"""Registry for health checker types."""
|
|
|
|
def __init__(self):
|
|
"""Initialize the registry with default checkers."""
|
|
self._checkers: Dict[str, Type[BaseHealthChecker]] = {
|
|
"api": APIHealthChecker,
|
|
"sensor": SensorHealthChecker,
|
|
"custom": CustomHealthChecker,
|
|
}
|
|
logger.debug(f"Initialized health checker registry with {len(self._checkers)} checkers")
|
|
|
|
def register(self, name: str, checker_class: Type[BaseHealthChecker]) -> None:
|
|
"""
|
|
Register a new health checker type.
|
|
|
|
Args:
|
|
name: Name of the checker type
|
|
checker_class: Health checker class
|
|
"""
|
|
self._checkers[name] = checker_class
|
|
logger.info(f"Registered health checker: {name}")
|
|
|
|
def get_checker(self, name: str) -> Type[BaseHealthChecker]:
|
|
"""
|
|
Get a health checker class by name.
|
|
|
|
Args:
|
|
name: Name of the checker type
|
|
|
|
Returns:
|
|
Health checker class
|
|
|
|
Raises:
|
|
ValueError: If checker type not found
|
|
"""
|
|
if name not in self._checkers:
|
|
available = ", ".join(self._checkers.keys())
|
|
raise ValueError(f"Unknown health checker type '{name}'. Available: {available}")
|
|
|
|
return self._checkers[name]
|
|
|
|
def list_checkers(self) -> list[str]:
|
|
"""
|
|
List all available health checker types.
|
|
|
|
Returns:
|
|
List of checker type names
|
|
"""
|
|
return list(self._checkers.keys())
|
|
|
|
|
|
class HealthCheckerFactory:
|
|
"""Factory for creating health checker instances."""
|
|
|
|
def __init__(self, registry: Optional[HealthCheckerRegistry] = None):
|
|
"""
|
|
Initialize the factory.
|
|
|
|
Args:
|
|
registry: Health checker registry (uses default if None)
|
|
"""
|
|
self.registry = registry or HealthCheckerRegistry()
|
|
logger.debug("Initialized health checker factory")
|
|
|
|
def create_checker(self, checker_type: str, timeout: float = 5.0) -> BaseHealthChecker:
|
|
"""
|
|
Create a health checker instance.
|
|
|
|
Args:
|
|
checker_type: Type of checker to create
|
|
timeout: Request timeout in seconds
|
|
|
|
Returns:
|
|
Health checker instance
|
|
"""
|
|
checker_class = self.registry.get_checker(checker_type)
|
|
checker = checker_class(timeout=timeout)
|
|
logger.debug(f"Created {checker_type} health checker with timeout {timeout}s")
|
|
return checker
|
|
|
|
def create_checker_for_service(self, service_name: str, config: Dict[str, Any], timeout: float = 5.0) -> BaseHealthChecker:
|
|
"""
|
|
Create a health checker for a specific service based on its configuration.
|
|
|
|
Args:
|
|
service_name: Name of the service
|
|
config: Service configuration
|
|
timeout: Request timeout in seconds
|
|
|
|
Returns:
|
|
Health checker instance
|
|
"""
|
|
# Determine checker type from config
|
|
checker_type = config.get("health_check_type", "api")
|
|
|
|
# Override based on service-specific logic
|
|
if service_name == "home_assistant" and config.get("sensor_entity"):
|
|
checker_type = "sensor"
|
|
elif config.get("health_checks"):
|
|
checker_type = "custom"
|
|
|
|
logger.debug(f"Creating {checker_type} checker for {service_name}")
|
|
return self.create_checker(checker_type, timeout)
|
|
|
|
|
|
# Global registry and factory instances
|
|
registry = HealthCheckerRegistry()
|
|
factory = HealthCheckerFactory(registry)
|