fix: Clean up whitespace and improve code formatting across service adapters
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
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.
This commit is contained in:
@@ -5,7 +5,7 @@ This module provides a registry and factory for different health checker types.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from typing import Any, Dict, Type
|
||||
from typing import Any, Dict, Optional, Type
|
||||
|
||||
from .api_checker import APIHealthChecker
|
||||
from .base import BaseHealthChecker
|
||||
@@ -17,7 +17,7 @@ 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]] = {
|
||||
@@ -30,7 +30,7 @@ class HealthCheckerRegistry:
|
||||
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
|
||||
@@ -41,26 +41,26 @@ class HealthCheckerRegistry:
|
||||
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
|
||||
"""
|
||||
@@ -69,29 +69,25 @@ class HealthCheckerRegistry:
|
||||
|
||||
class HealthCheckerFactory:
|
||||
"""Factory for creating health checker instances."""
|
||||
|
||||
def __init__(self, registry: HealthCheckerRegistry = None):
|
||||
|
||||
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:
|
||||
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
|
||||
"""
|
||||
@@ -100,32 +96,27 @@ class HealthCheckerFactory:
|
||||
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:
|
||||
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)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user