add healthchecks client

src/lego_monitoring/alerting/alert.py

@@ -1,5 +1,6 @@
 from dataclasses import dataclass, field
 from datetime import datetime
+from typing import Optional
 
 from .enum import AlertType, Severity
 
@@ -10,3 +11,9 @@ class Alert:
     message: str
     severity: Severity
     created: datetime = field(default_factory=datetime.now)
+    healthchecks_slug: Optional[str] = None
+    plain_message: Optional[str] = None
+
+    def __post_init__(self):
+        if self.plain_message is None:
+            self.plain_message = self.message

src/lego_monitoring/alerting/clients/common.py

@@ -0,0 +1,10 @@
+class UnsuccessfulRequest(Exception): ...
+
+
+def raise_for_status(response):
+    """Checks whether or not the response was successful."""
+    if 200 <= response.status_code < 300:
+        # Pass through the response.
+        return response
+
+    raise UnsuccessfulRequest(response.url)

src/lego_monitoring/alerting/clients/healthchecks.py

@@ -0,0 +1,20 @@
+from typing import Optional
+
+from uplink import Body, Consumer, Path, Query, post, response_handler
+
+from .common import raise_for_status
+
+
+@response_handler(raise_for_status)
+class HealthchecksClient(Consumer):
+    @post("{key}/{slug}")
+    def _success(self, key: Path, slug: Path, create: Query, log: Body): ...
+
+    @post("{key}/{slug}/fail")
+    def _failure(self, key: Path, slug: Path, create: Query, log: Body): ...
+
+    def success(self, key: Path, slug: str, create: bool = False, log: Optional[str] = None):
+        return self._success(key, slug, int(create), log)
+
+    def failure(self, key: Path, slug: str, create: bool = False, log: Optional[str] = None):
+        return self._failure(key, slug, int(create), log)

src/lego_monitoring/alerting/sender.py

@@ -1,12 +1,14 @@
 from telethon import TelegramClient
 from telethon.sessions import MemorySession
+from uplink import AiohttpClient
 
 from ..core import cvars
 from .alert import Alert
+from .clients.healthchecks import HealthchecksClient
 from .enum import SEVERITY_TO_EMOJI, AlertType, Severity
 
 
-async def get_client() -> TelegramClient:
+async def get_tg_client() -> TelegramClient:
     config = cvars.config.get()
     api_id, api_hash, bot_token = config.alert_channels.telegram.creds.split(",")
     client = await TelegramClient(MemorySession(), api_id, api_hash, connection_retries=None).start(bot_token=bot_token)
@@ -14,6 +16,15 @@ async def get_client() -> TelegramClient:
     return client
 
 
+def get_healthchecks_client() -> HealthchecksClient:
+    config = cvars.config.get()
+    base_url = config.alert_channels.healthchecks.pinging_api_endpoint
+    client = HealthchecksClient(
+        base_url=config.alert_channels.healthchecks.pinging_api_endpoint, client=AiohttpClient()
+    )
+    return client
+
+
 def format_message(alert: Alert, note: str) -> str:
     severity_emoji = SEVERITY_TO_EMOJI[alert.severity]
     note_formatted = f"{note}, " if note else ""
@@ -31,7 +42,7 @@ async def send_alert(alert: Alert, note: str = "") -> None:
     except LookupError:  # being called standalone
         # cvars.config.set(get_config())
         # temp_client = True
-        # client = await get_client()
+        # client = await get_tg_client()
         # cvars.matrix_client.set(client)
         raise NotImplementedError  # TODO
     else:
@@ -42,7 +53,10 @@ async def send_alert(alert: Alert, note: str = "") -> None:
         await client.send_message(entity=room_id, message=message)
         # if temp_client:
         #     await client.close()
+
     # TODO ping healthchecks if enabled
+    if alert.healthchecks_slug is not None:
+        raise NotImplementedError
 
 
 # TODO service itself has to be monitored like everything else - with regular pinging - if we're