64 lines
1.6 KiB
Python
64 lines
1.6 KiB
Python
from dataclasses import dataclass
|
|
from datetime import timedelta
|
|
from typing import Awaitable, Callable, Protocol
|
|
|
|
from ..agent import AgentResult, CheckmkReader
|
|
from .models import Command, IncomingMessage, Reply
|
|
|
|
|
|
class AgentPort(Protocol):
|
|
async def ask(self, question: str) -> AgentResult: ...
|
|
|
|
async def explain(self, host: str, service: str) -> AgentResult: ...
|
|
|
|
|
|
class AlertLookup(Protocol):
|
|
def services_for_message(self, backend: str, target: str, message_id: str) -> list[tuple[str, str]]: ...
|
|
|
|
def service_for_alert(self, alert_id: int) -> tuple[str, str] | None: ...
|
|
|
|
|
|
class MutePort(Protocol):
|
|
def create(self, host: str, service: str, duration: timedelta, reason: str, created_by: str, source: str): ...
|
|
|
|
|
|
class DowntimePort(Protocol):
|
|
def enabled(self) -> bool: ...
|
|
|
|
def max_duration(self) -> timedelta: ...
|
|
|
|
async def request(self, backend: str, sender_id: str, host: str, service: str, duration: timedelta,
|
|
comment: str): ...
|
|
|
|
async def resolve(self, token: str, backend: str, sender_id: str, approve: bool): ...
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CommandDeps:
|
|
reader: CheckmkReader
|
|
agent: AgentPort
|
|
mutes: MutePort | None = None
|
|
lookup: AlertLookup | None = None
|
|
downtime: DowntimePort | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CommandContext:
|
|
message: IncomingMessage
|
|
command: Command
|
|
deps: CommandDeps
|
|
registry: "Registry"
|
|
|
|
|
|
Handler = Callable[[CommandContext], Awaitable[Reply]]
|
|
Registry = dict[str, Handler]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AuditEvent:
|
|
backend: str
|
|
sender_id: str
|
|
command: str
|
|
args: str
|
|
outcome: str
|