82 lines
3.1 KiB
Python
82 lines
3.1 KiB
Python
from datetime import datetime, timezone
|
|
|
|
import httpx
|
|
|
|
from ..checkmk import API_PATH, DEFAULT_TIMEOUT, VERSION_PATH
|
|
from ..chat import mask_secret
|
|
from ..constants import (CHECKMK_CA_FILE, CHECKMK_DT_SECRET, CHECKMK_DT_USER, CHECKMK_SITE, CHECKMK_URL,
|
|
CHECKMK_VERIFY_TLS, FLAG_OFF, FLAG_ON)
|
|
|
|
DOWNTIME_PATH = "/domain-types/downtime/collections/service"
|
|
TIME_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
|
|
ERROR_MAX = 200
|
|
|
|
|
|
class DowntimeError(Exception):
|
|
pass
|
|
|
|
|
|
def iso_utc(moment: datetime) -> str:
|
|
return moment.astimezone(timezone.utc).strftime(TIME_FORMAT)
|
|
|
|
|
|
def is_configured(store) -> bool:
|
|
return all(store.get(key) for key in (CHECKMK_URL, CHECKMK_SITE, CHECKMK_DT_USER, CHECKMK_DT_SECRET))
|
|
|
|
|
|
class CheckmkDowntimeClient:
|
|
def __init__(self, base_url: str, site: str, user: str, secret: str, verify: bool | str = True,
|
|
timeout: float = DEFAULT_TIMEOUT, transport: httpx.AsyncBaseTransport | None = None):
|
|
self._secret = secret
|
|
self._client = httpx.AsyncClient(
|
|
base_url=f"{base_url.rstrip('/')}/{site}{API_PATH}",
|
|
headers={"Authorization": f"Bearer {user} {secret}", "Accept": "application/json"},
|
|
verify=verify, timeout=timeout, transport=transport,
|
|
)
|
|
|
|
@classmethod
|
|
def from_settings(cls, store) -> "CheckmkDowntimeClient | None":
|
|
if not is_configured(store):
|
|
return None
|
|
if store.get(CHECKMK_VERIFY_TLS, FLAG_ON) == FLAG_OFF:
|
|
verify: bool | str = False
|
|
else:
|
|
verify = store.get(CHECKMK_CA_FILE) or True
|
|
return cls(store.get(CHECKMK_URL), store.get(CHECKMK_SITE), store.get(CHECKMK_DT_USER),
|
|
store.get(CHECKMK_DT_SECRET), verify=verify)
|
|
|
|
def _mask(self, text: str) -> str:
|
|
return mask_secret(text, self._secret)[:ERROR_MAX]
|
|
|
|
async def _request(self, method: str, path: str, **kwargs) -> httpx.Response:
|
|
try:
|
|
response = await self._client.request(method, path, **kwargs)
|
|
except httpx.HTTPError as exc:
|
|
raise DowntimeError(self._mask(f"{type(exc).__name__}: {exc}")) from None
|
|
if response.status_code >= 400:
|
|
try:
|
|
body = response.json()
|
|
detail = str(body.get("detail") or body.get("title") or "")
|
|
except (ValueError, AttributeError):
|
|
detail = ""
|
|
raise DowntimeError(self._mask(f"HTTP {response.status_code}: {detail}".strip()))
|
|
return response
|
|
|
|
async def version(self) -> dict:
|
|
return (await self._request("GET", VERSION_PATH)).json()
|
|
|
|
async def create_service_downtime(self, host: str, service: str, start: datetime, end: datetime,
|
|
comment: str) -> None:
|
|
payload = {
|
|
"downtime_type": "service",
|
|
"host_name": host,
|
|
"service_descriptions": [service],
|
|
"start_time": iso_utc(start),
|
|
"end_time": iso_utc(end),
|
|
"comment": comment,
|
|
}
|
|
await self._request("POST", DOWNTIME_PATH, json=payload)
|
|
|
|
async def aclose(self) -> None:
|
|
await self._client.aclose()
|