42 lines
1.9 KiB
Python
42 lines
1.9 KiB
Python
import inspect
|
|
|
|
from .constants import (TOOL_DOWNTIMES, TOOL_HOST_SERVICES, TOOL_PROBLEMS, TOOL_SERVICE_STATUS,
|
|
TOOL_RESULT_MAX, TOOL_STATE_HISTORY)
|
|
from .formatting import (format_downtimes, format_history, format_services, format_status)
|
|
from .sanitize import sanitize
|
|
from .tools import ToolArgumentError, validate_arguments
|
|
from .types import CheckmkReader, StateLog
|
|
|
|
|
|
class ToolExecutor:
|
|
def __init__(self, reader: CheckmkReader, state_log: StateLog):
|
|
self._reader = reader
|
|
self._state_log = state_log
|
|
|
|
async def _dispatch(self, tool: str, args: dict[str, str]) -> str:
|
|
if tool == TOOL_SERVICE_STATUS:
|
|
return format_status(await self._reader.service_status(args["host"], args["service"]))
|
|
if tool == TOOL_HOST_SERVICES:
|
|
return format_services(await self._reader.host_services(args["host"]), with_host=False)
|
|
if tool == TOOL_STATE_HISTORY:
|
|
changes = self._state_log(args["host"], args["service"])
|
|
if inspect.isawaitable(changes):
|
|
changes = await changes
|
|
return format_history(changes)
|
|
if tool == TOOL_DOWNTIMES:
|
|
return format_downtimes(await self._reader.downtimes(args.get("host")))
|
|
if tool == TOOL_PROBLEMS:
|
|
return format_services(await self._reader.problems())
|
|
raise ToolArgumentError("unbekanntes Tool")
|
|
|
|
async def execute(self, tool: str, arguments: object) -> tuple[str, str | None]:
|
|
"""Liefert (Ergebnistext, Fehler); wirft nie."""
|
|
try:
|
|
args = validate_arguments(tool, arguments)
|
|
except ToolArgumentError as exc:
|
|
return f"Fehler: {exc}", str(exc)
|
|
try:
|
|
return sanitize(await self._dispatch(tool, args), TOOL_RESULT_MAX), None
|
|
except Exception as exc:
|
|
return "Fehler: Abfrage fehlgeschlagen.", type(exc).__name__
|