38 lines
1.3 KiB
Python
38 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
import httpx
|
|
|
|
from ...redaction import redact_dict
|
|
from ...timeutil import payload_with_display_times
|
|
from .base import DeliveryResult
|
|
from .http import classify_response
|
|
|
|
|
|
class WebhookNotifier:
|
|
name = "webhook"
|
|
|
|
def __init__(
|
|
self, client: httpx.AsyncClient, url: str, token: str = "", time_zone: str = "UTC"
|
|
) -> None:
|
|
self._client = client
|
|
self._url = url
|
|
self._token = token
|
|
self._time_zone = time_zone
|
|
|
|
async def deliver(self, event_type: str, payload: dict) -> DeliveryResult:
|
|
headers = {"Content-Type": "application/json"}
|
|
if self._token:
|
|
headers["Authorization"] = f"Bearer {self._token}"
|
|
body = {
|
|
"event_type": event_type,
|
|
"incident": redact_dict(payload_with_display_times(payload, self._time_zone)),
|
|
}
|
|
try:
|
|
resp = await self._client.post(self._url, json=body, headers=headers)
|
|
except httpx.HTTPError as exc:
|
|
return DeliveryResult(ok=False, permanent=False, error=type(exc).__name__)
|
|
ok, perm = classify_response(resp.status_code)
|
|
if ok:
|
|
return DeliveryResult(ok=True)
|
|
return DeliveryResult(ok=False, permanent=perm, error=f"http_{resp.status_code}")
|