Coverage for src/lexigram/notification/backends/slack/slack_notifier.py: 94%
108 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 02:32 +0800
1"""Slack notification backend.
3Supports two delivery modes:
5- **Webhook** — posts a message to a Slack Incoming Webhook URL via
6 ``httpx``. No dependency on the Slack SDK.
7- **Bot API** — uses the ``slack-sdk`` ``AsyncWebClient`` to send rich
8 messages, attach blocks, target specific channels, and reply in threads.
9 Requires the ``lexigram-notification[slack]`` extra.
10"""
12from __future__ import annotations
14from typing import Any
15import uuid
17from lexigram.contracts.core import HealthCheckResult, HealthStatus
18from lexigram.contracts.mailer import MessageDeliveryReceipt
19from lexigram.contracts.notification.errors import NotificationError
20from lexigram.logging import get_logger
21from lexigram.result import Err, Ok, Result
23logger = get_logger(__name__)
25_DEFAULT_TIMEOUT = 30
27_MRKDWN_ESCAPES = str.maketrans({"<": "<", ">": ">", "&": "&"})
30def _escape_mrkdwn(text: str) -> str:
31 """Escape Slack mrkdwn-special characters in user-supplied text.
33 Slack interprets ``<url|label>`` link markup and ``&`` entities in the
34 ``text`` field. Apply before sending untrusted content so it renders as
35 literal text instead of a spoofable link.
37 Args:
38 text: Message text.
40 Returns:
41 Text with ``<``, ``>``, and ``&`` escaped.
42 """
43 return text.translate(_MRKDWN_ESCAPES)
46class SlackNotificationError(NotificationError):
47 """Slack notification delivery failure."""
49 _code = "LEX_ERR_NOTIF_014"
51 def __init__(
52 self,
53 message: str = "Slack delivery error",
54 *,
55 slack_error: str | None = None,
56 ) -> None:
57 super().__init__(message, channel="chat", backend="slack")
58 self.slack_error = slack_error
61class SlackMessage:
62 """A message to be delivered via Slack.
64 Args:
65 text: Plaintext fallback message body.
66 channel: Target channel or user ID (Bot API only). Overrides
67 the ``default_channel`` set on the backend.
68 blocks: Optional Block Kit blocks for rich formatting.
69 thread_ts: Optional parent message timestamp for threaded replies.
70 """
72 __slots__ = ("blocks", "channel", "text", "thread_ts")
74 def __init__(
75 self,
76 text: str,
77 *,
78 channel: str | None = None,
79 blocks: list[dict[str, Any]] | None = None,
80 thread_ts: str | None = None,
81 ) -> None:
82 self.text = text
83 self.channel = channel
84 self.blocks = blocks
85 self.thread_ts = thread_ts
88class SlackNotifier:
89 """Slack notification backend.
91 Supports two delivery modes:
93 - **Incoming Webhook** (``mode="webhook"``): Posts a message to a
94 pre-configured webhook URL. Simple and credential-free; does not
95 require ``slack-sdk``. Uses ``httpx``.
96 - **Bot API** (``mode="bot"``): Uses the ``slack-sdk``
97 ``AsyncWebClient`` to send to any channel or user, with support
98 for Block Kit, threading, and other rich features. Requires the
99 ``lexigram-notification[slack]`` extra.
101 Args:
102 webhook_url: Slack Incoming Webhook URL (webhook mode).
103 bot_token: Slack Bot OAuth token beginning with ``xoxb-``
104 (bot mode).
105 default_channel: Default channel / user to message when
106 :class:`SlackMessage` does not specify one (bot mode).
107 timeout: HTTP request timeout in seconds.
108 """
110 def __init__(
111 self,
112 *,
113 webhook_url: str | None = None,
114 bot_token: str | None = None,
115 default_channel: str | None = None,
116 timeout: int = _DEFAULT_TIMEOUT,
117 ) -> None:
118 if not webhook_url and not bot_token:
119 raise ValueError(
120 "SlackNotifier requires either webhook_url (webhook mode) "
121 "or bot_token (bot mode)."
122 )
123 self._webhook_url = webhook_url
124 self._bot_token = bot_token
125 self._default_channel = default_channel
126 self._timeout = timeout
128 @property
129 def _mode(self) -> str:
130 """Active delivery mode."""
131 return "webhook" if self._webhook_url else "bot"
133 # ──────────────────────────────────────────────────────────────
134 # Public interface
135 # ──────────────────────────────────────────────────────────────
137 async def send(
138 self, message: SlackMessage
139 ) -> Result[MessageDeliveryReceipt, NotificationError]:
140 """Send a Slack message.
142 Dispatches to the webhook or Bot API implementation depending on
143 which credentials were supplied at construction time.
145 Args:
146 message: :class:`SlackMessage` to deliver.
148 Returns:
149 ``Ok(MessageDeliveryReceipt)`` on success.
150 ``Err(SlackNotificationError)`` on delivery failure.
151 """
152 if self._mode == "webhook":
153 return await self._send_webhook(message)
154 return await self._send_bot(message)
156 async def send_text(
157 self,
158 text: str,
159 *,
160 channel: str | None = None,
161 thread_ts: str | None = None,
162 ) -> Result[MessageDeliveryReceipt, NotificationError]:
163 """Convenience wrapper that sends a plain-text Slack message.
165 Args:
166 text: Message body.
167 channel: Target channel (bot mode only).
168 thread_ts: Thread parent timestamp for threaded replies.
170 Returns:
171 ``Ok(MessageDeliveryReceipt)`` on success.
172 """
173 return await self.send(SlackMessage(text, channel=channel, thread_ts=thread_ts))
175 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult:
176 """Check Slack API reachability.
178 - Webhook mode: performs a GET to ``https://slack.com``.
179 - Bot API mode: calls ``auth.test`` with the bot token.
181 Args:
182 timeout: Max seconds to wait.
184 Returns:
185 :class:`~lexigram.contracts.core.HealthCheckResult`.
186 """
187 if self._mode == "webhook":
188 return await self._health_webhook(timeout)
189 return await self._health_bot(timeout)
191 # ──────────────────────────────────────────────────────────────
192 # Webhook mode
193 # ──────────────────────────────────────────────────────────────
195 async def _send_webhook(
196 self, message: SlackMessage
197 ) -> Result[MessageDeliveryReceipt, NotificationError]:
198 """Post to a Slack Incoming Webhook URL via httpx."""
199 try:
200 import httpx
201 except ImportError as exc:
202 raise ImportError(
203 "httpx is required for SlackNotifier webhook mode. "
204 "Install with: pip install lexigram-notification[slack]"
205 ) from exc
207 payload: dict[str, Any] = {"text": _escape_mrkdwn(message.text)}
208 if message.blocks:
209 payload["blocks"] = message.blocks
211 async with httpx.AsyncClient(timeout=self._timeout) as client:
212 resp = await client.post(
213 self._webhook_url, # type: ignore[arg-type]
214 json=payload,
215 )
217 # Slack webhooks return 200 with plain-text ``"ok"`` on success.
218 if resp.status_code == 200 and resp.text.strip() == "ok":
219 receipt = MessageDeliveryReceipt(
220 message_id=str(uuid.uuid4()),
221 backend="slack_webhook",
222 channel="slack",
223 )
224 logger.info("slack.webhook_sent", text_preview=message.text[:50])
225 return Ok(receipt)
227 error_text = resp.text.strip() or f"HTTP {resp.status_code}"
228 logger.warning(
229 "slack.webhook_send_failed",
230 status=resp.status_code,
231 body=error_text,
232 )
233 return Err(
234 SlackNotificationError(
235 f"Slack webhook error: {error_text}",
236 slack_error=error_text,
237 )
238 )
240 async def _health_webhook(self, timeout: float) -> HealthCheckResult:
241 """Lightweight connectivity check for webhook mode."""
242 try:
243 import httpx
245 async with httpx.AsyncClient(timeout=timeout) as client:
246 resp = await client.get("https://slack.com")
247 status = (
248 HealthStatus.HEALTHY
249 if resp.status_code < 500
250 else HealthStatus.UNHEALTHY
251 )
252 return HealthCheckResult(
253 component="slack_webhook",
254 status=status,
255 details={"http_status": resp.status_code},
256 )
257 except OSError as exc:
258 return HealthCheckResult(
259 component="slack_webhook",
260 status=HealthStatus.UNHEALTHY,
261 details={"error": str(exc)},
262 )
264 # ──────────────────────────────────────────────────────────────
265 # Bot API mode
266 # ──────────────────────────────────────────────────────────────
268 async def _send_bot(
269 self, message: SlackMessage
270 ) -> Result[MessageDeliveryReceipt, NotificationError]:
271 """Send via slack-sdk AsyncWebClient."""
272 try:
273 from slack_sdk.web.async_client import ( # type: ignore[import-not-found]
274 AsyncWebClient,
275 )
276 except ImportError as exc:
277 raise ImportError(
278 "slack-sdk is required for SlackNotifier bot mode. "
279 "Install with: pip install lexigram-notification[slack]"
280 ) from exc
282 channel = message.channel or self._default_channel
283 if not channel:
284 return Err(
285 SlackNotificationError(
286 "SlackNotifier bot mode requires a channel. "
287 "Set default_channel or pass message.channel.",
288 slack_error="missing_channel",
289 )
290 )
292 client = AsyncWebClient(token=self._bot_token)
293 post_kwargs: dict[str, Any] = {
294 "channel": channel,
295 "text": _escape_mrkdwn(message.text),
296 }
297 if message.blocks:
298 post_kwargs["blocks"] = message.blocks
299 if message.thread_ts:
300 post_kwargs["thread_ts"] = message.thread_ts
302 resp = await client.chat_postMessage(**post_kwargs)
303 if resp["ok"]:
304 receipt = MessageDeliveryReceipt(
305 message_id=str(uuid.uuid4()),
306 backend="slack_bot",
307 channel="slack",
308 provider_reference=resp.get("ts"),
309 )
310 logger.info(
311 "slack.bot_sent",
312 channel=channel,
313 ts=resp.get("ts"),
314 text_preview=message.text[:50],
315 )
316 return Ok(receipt)
318 error_code = resp.get("error", "unknown")
319 logger.warning("slack.bot_send_failed", channel=channel, error=error_code)
320 return Err(
321 SlackNotificationError(
322 f"Slack bot API error: {error_code}",
323 slack_error=error_code,
324 )
325 )
327 async def _health_bot(self, timeout: float) -> HealthCheckResult:
328 """Health check via Slack ``auth.test`` API."""
329 try:
330 from slack_sdk.web.async_client import (
331 AsyncWebClient,
332 )
333 except ImportError:
334 return HealthCheckResult(
335 component="slack_bot",
336 status=HealthStatus.UNHEALTHY,
337 message="slack-sdk is not installed",
338 )
340 try:
341 client = AsyncWebClient(token=self._bot_token, timeout=timeout)
342 resp = await client.auth_test()
343 if resp["ok"]:
344 return HealthCheckResult(
345 component="slack_bot",
346 status=HealthStatus.HEALTHY,
347 details={"team": resp.get("team"), "bot_id": resp.get("bot_id")},
348 )
349 return HealthCheckResult(
350 component="slack_bot",
351 status=HealthStatus.UNHEALTHY,
352 message=f"auth.test failed: {resp.get('error')}",
353 )
354 except OSError as exc:
355 return HealthCheckResult(
356 component="slack_bot",
357 status=HealthStatus.UNHEALTHY,
358 details={"error": str(exc)},
359 )
362__all__ = [
363 "SlackMessage",
364 "SlackNotificationError",
365 "SlackNotifier",
366]