79 lines
2.2 KiB
Python
79 lines
2.2 KiB
Python
"""
|
|
Monitoring command registry for extensible command discovery and execution.
|
|
"""
|
|
import logging
|
|
from typing import Dict, Optional, Type
|
|
|
|
from core.services.monitoring.base import BaseMonitoringCommand, MonitoringResult
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class MonitoringCommandRegistry:
|
|
"""
|
|
Singleton registry for monitoring commands.
|
|
|
|
Usage:
|
|
# Register a command
|
|
@MonitoringCommandRegistry.register
|
|
class MyCommand(BaseMonitoringCommand):
|
|
...
|
|
|
|
# Execute a command
|
|
result = MonitoringCommandRegistry.execute('my_command')
|
|
"""
|
|
_commands: Dict[str, Type[BaseMonitoringCommand]] = {}
|
|
|
|
@classmethod
|
|
def register(cls, command_class: Type[BaseMonitoringCommand]) -> Type[BaseMonitoringCommand]:
|
|
"""
|
|
Register a monitoring command class.
|
|
Can be used as a decorator.
|
|
|
|
Args:
|
|
command_class: BaseMonitoringCommand subclass
|
|
|
|
Returns:
|
|
The command class (unchanged)
|
|
"""
|
|
instance = command_class()
|
|
cls._commands[instance.name] = command_class
|
|
logger.info(f"Registered monitoring command: {instance.name}")
|
|
return command_class
|
|
|
|
@classmethod
|
|
def get_command(cls, name: str) -> Optional[Type[BaseMonitoringCommand]]:
|
|
"""Get a command class by name."""
|
|
return cls._commands.get(name)
|
|
|
|
@classmethod
|
|
def execute(cls, name: str, **kwargs) -> MonitoringResult:
|
|
"""
|
|
Execute a registered command by name.
|
|
|
|
Args:
|
|
name: Command name
|
|
**kwargs: Arguments passed to command execute()
|
|
|
|
Returns:
|
|
MonitoringResult from command execution
|
|
|
|
Raises:
|
|
ValueError: If command not found
|
|
"""
|
|
command_class = cls.get_command(name)
|
|
if not command_class:
|
|
raise ValueError(f"Monitoring command not found: {name}")
|
|
|
|
command = command_class()
|
|
logger.info(f"Executing monitoring command: {name}")
|
|
return command.execute(**kwargs)
|
|
|
|
@classmethod
|
|
def list_commands(cls) -> Dict[str, str]:
|
|
"""List all registered commands with descriptions."""
|
|
return {
|
|
name: cls._commands[name]().description
|
|
for name in cls._commands
|
|
}
|