35 lines
1.1 KiB
Python
35 lines
1.1 KiB
Python
from typing import Callable
|
|
|
|
from ..checkmk import CheckmkClient, Downtime, ServiceState
|
|
|
|
|
|
class CheckmkNotConfigured(Exception):
|
|
pass
|
|
|
|
|
|
class SettingsReader:
|
|
def __init__(self, store, factory: Callable[[object], CheckmkClient | None] = CheckmkClient.from_settings):
|
|
self._store = store
|
|
self._factory = factory
|
|
|
|
async def _call(self, method: str, *args):
|
|
client = self._factory(self._store)
|
|
if client is None:
|
|
raise CheckmkNotConfigured
|
|
try:
|
|
return await getattr(client, method)(*args)
|
|
finally:
|
|
await client.aclose()
|
|
|
|
async def service_status(self, host: str, service: str) -> ServiceState | None:
|
|
return await self._call("service_status", host, service)
|
|
|
|
async def host_services(self, host: str) -> list[ServiceState]:
|
|
return await self._call("host_services", host)
|
|
|
|
async def downtimes(self, host: str | None = None) -> list[Downtime]:
|
|
return await self._call("downtimes", host)
|
|
|
|
async def problems(self) -> list[ServiceState]:
|
|
return await self._call("problems")
|