Source code for corvix.notifications.targets.slack
"""Slack incoming-webhook notification target.
This is the built-in reference :class:`~corvix.notifications.targets.base.NotificationTarget`
implementation. It posts one Slack message per newly-arrived unread GitHub
notification to a configured incoming-webhook URL. See
``docs/new_notification_target.md`` for how to add your own channel.
"""
from __future__ import annotations
import json
import logging
import urllib.request
from dataclasses import dataclass
from corvix.notifications.models import DeliveryResult, NotificationEvent
[docs]
logger = logging.getLogger(__name__)
[docs]
_DEFAULT_TIMEOUT_SECONDS = 5.0
@dataclass(slots=True)
[docs]
class SlackTarget:
"""Posts a Slack message for each new GitHub notification.
Parameters
----------
webhook_url:
Slack incoming-webhook URL. Resolved from an env var by the CLI so the
secret never lives in ``corvix.yaml``.
enabled:
When ``False`` the target is a no-op (delivers nothing). The CLI already
skips disabled targets, but this keeps the target safe to construct
directly.
timeout_seconds:
Per-request HTTP timeout.
"""
[docs]
timeout_seconds: float = _DEFAULT_TIMEOUT_SECONDS
@property
[docs]
def name(self) -> str:
return "slack"
[docs]
def deliver(self, events: list[NotificationEvent]) -> DeliveryResult:
"""Post one message per event; never raise (errors go in the result)."""
if not self.enabled:
return DeliveryResult(target=self.name, events_attempted=len(events), events_delivered=0)
errors: list[str] = []
delivered = 0
for event in events:
try:
self._post(_format_message(event))
except (OSError, ValueError) as exc:
# urllib.error.URLError / HTTPError are OSError subclasses.
logger.warning("Slack delivery failed for thread %s: %s", event.thread_id, exc)
errors.append(f"{event.thread_id}: {exc}")
else:
delivered += 1
return DeliveryResult(
target=self.name,
events_attempted=len(events),
events_delivered=delivered,
errors=errors,
)
[docs]
def _post(self, text: str) -> None:
payload = json.dumps({"text": text}).encode("utf-8")
request = urllib.request.Request(
self.webhook_url,
data=payload,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(request, timeout=self.timeout_seconds):
# A non-2xx status raises HTTPError (an OSError); the body is unused.
pass