Coverage for src/lexigram/notification/backends/sms/whatsapp.py: 97%
116 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 02:32 +0800
1"""WhatsApp Business API notification backend.
3Supports two delivery providers:
5- **Twilio** — uses the Twilio Messaging API with the ``whatsapp:`` URI
6 scheme. Requires the existing ``aiohttp`` dependency.
7- **Meta** — calls the Meta (Facebook) WhatsApp Business API directly
8 via ``httpx``. Requires the ``lexigram-notification[whatsapp]`` extra.
9"""
11from __future__ import annotations
13import base64
14from typing import TYPE_CHECKING, Any
15import uuid
17from lexigram.contracts.core import HealthCheckResult, HealthStatus
18from lexigram.contracts.mailer import MessageDeliveryReceipt
19from lexigram.contracts.notification.types import SMSMessage
20from lexigram.logging import get_logger
21from lexigram.notification.constants import DEFAULT_TWILIO_TIMEOUT, TWILIO_API_BASE
22from lexigram.notification.exceptions import TwilioNotificationError
23from lexigram.result import Err, Ok, Result
25if TYPE_CHECKING:
26 from lexigram.contracts.notification.errors import NotificationError
28logger = get_logger(__name__)
30# Meta Graph API base URL for WhatsApp Business messages.
31_META_API_BASE = "https://graph.facebook.com/v19.0"
32_DEFAULT_META_TIMEOUT = 30
35class WhatsAppNotificationError(TwilioNotificationError):
36 """WhatsApp delivery failure via Twilio provider."""
38 _code = "LEX_ERR_NOTIF_012"
40 def __init__(
41 self,
42 message: str = "WhatsApp (Twilio) delivery error",
43 *,
44 twilio_code: int | None = None,
45 **kwargs: Any,
46 ) -> None:
47 # Call NotificationError directly; super chain reaches TwilioNotificationError
48 # but we override backend to "whatsapp_twilio".
49 super().__init__(message, twilio_code=twilio_code, **kwargs)
50 self.backend = "whatsapp_twilio"
53class WhatsAppMetaNotificationError(TwilioNotificationError):
54 """WhatsApp delivery failure via Meta provider."""
56 _code = "LEX_ERR_NOTIF_013"
58 def __init__(
59 self,
60 message: str = "WhatsApp (Meta) delivery error",
61 *,
62 meta_code: str | None = None,
63 **kwargs: Any,
64 ) -> None:
65 super().__init__(message, **kwargs)
66 self.backend = "whatsapp_meta"
67 self.meta_code = meta_code
70class WhatsAppBackend:
71 """WhatsApp Business API notification backend.
73 Supports two providers:
75 - ``"twilio"`` — Twilio Messaging API with ``whatsapp:`` prefix.
76 Accepts the same ``account_sid`` / ``auth_token`` / ``from_number``
77 credentials as :class:`~lexigram.notification.backends.sms.twilio.TwilioSMS`.
78 - ``"meta"`` — Meta (Facebook) WhatsApp Business API.
79 Requires ``access_token`` and ``phone_number_id``.
81 Implements a compatible ``send`` / ``health_check`` interface.
83 Args:
84 provider: Delivery provider. Either ``"twilio"`` or ``"meta"``.
85 account_sid: Twilio Account SID (Twilio provider only).
86 auth_token: Twilio Auth Token (Twilio provider only).
87 from_number: Twilio WhatsApp-enabled number in ``whatsapp:+1…``
88 format, or just the E.164 number (prefix added automatically).
89 access_token: Meta Graph API access token (Meta provider only).
90 phone_number_id: Meta phone number object ID (Meta provider only).
91 timeout: HTTP request timeout in seconds.
92 """
94 def __init__(
95 self,
96 *,
97 provider: str = "twilio",
98 account_sid: str | None = None,
99 auth_token: str | None = None,
100 from_number: str | None = None,
101 access_token: str | None = None,
102 phone_number_id: str | None = None,
103 timeout: int = DEFAULT_TWILIO_TIMEOUT,
104 ) -> None:
105 if provider not in ("twilio", "meta"):
106 raise ValueError(
107 f"Unsupported WhatsApp provider: {provider!r}. "
108 "Expected 'twilio' or 'meta'."
109 )
110 self._provider = provider
111 self._account_sid = account_sid
112 self._auth_token = auth_token
113 self._from_number = from_number
114 self._access_token = access_token
115 self._phone_number_id = phone_number_id
116 self._timeout = timeout
118 # ──────────────────────────────────────────────────────────────
119 # Public interface
120 # ──────────────────────────────────────────────────────────────
122 async def send(
123 self, message: SMSMessage
124 ) -> Result[MessageDeliveryReceipt, NotificationError]:
125 """Send a WhatsApp message.
127 Dispatches to the appropriate provider implementation.
129 Args:
130 message: The :class:`~lexigram.contracts.notification.types.SMSMessage`
131 to deliver. ``message.to`` must contain the recipient phone
132 numbers in E.164 format.
134 Returns:
135 ``Ok(MessageDeliveryReceipt)`` on success.
136 ``Err(NotificationError)`` on delivery failure.
137 """
138 if self._provider == "twilio":
139 return await self._send_twilio(message)
140 return await self._send_meta(message)
142 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult:
143 """Check provider API reachability.
145 Args:
146 timeout: Max seconds to wait.
148 Returns:
149 :class:`~lexigram.contracts.core.HealthCheckResult`.
150 """
151 if self._provider == "twilio":
152 return await self._health_twilio(timeout)
153 return await self._health_meta(timeout)
155 # ──────────────────────────────────────────────────────────────
156 # Twilio provider
157 # ──────────────────────────────────────────────────────────────
159 def _twilio_auth_header(self) -> str:
160 """Build Basic Auth header for Twilio."""
161 credentials = f"{self._account_sid}:{self._auth_token}"
162 return "Basic " + base64.b64encode(credentials.encode()).decode()
164 def _whatsapp_number(self, number: str) -> str:
165 """Ensure the number has the ``whatsapp:`` prefix."""
166 if number.startswith("whatsapp:"):
167 return number
168 return f"whatsapp:{number}"
170 async def _send_twilio(
171 self, message: SMSMessage
172 ) -> Result[MessageDeliveryReceipt, NotificationError]:
173 """Send via Twilio Messaging API."""
174 import aiohttp
176 url = f"{TWILIO_API_BASE}/Accounts/{self._account_sid}/Messages.json"
177 headers = {
178 "Authorization": self._twilio_auth_header(),
179 "Content-Type": "application/x-www-form-urlencoded",
180 }
181 from_wa = self._whatsapp_number(message.from_number or self._from_number or "")
182 to_number = message.to[0] if message.to else ""
183 data = {
184 "To": self._whatsapp_number(to_number),
185 "From": from_wa,
186 "Body": message.body,
187 }
189 async with aiohttp.ClientSession() as session:
190 async with session.post(
191 url,
192 data=data,
193 headers=headers,
194 timeout=aiohttp.ClientTimeout(total=self._timeout),
195 ) as resp:
196 body = await resp.json()
197 if resp.status == 201:
198 receipt = MessageDeliveryReceipt(
199 message_id=str(uuid.uuid4()),
200 backend="whatsapp_twilio",
201 channel="whatsapp",
202 provider_reference=body.get("sid"),
203 )
204 logger.info(
205 "whatsapp.twilio_sent",
206 to=message.to,
207 sid=body.get("sid"),
208 )
209 return Ok(receipt)
211 error_msg = body.get("message", f"HTTP {resp.status}")
212 twilio_code = body.get("code")
213 logger.warning(
214 "whatsapp.twilio_send_failed",
215 status=resp.status,
216 code=twilio_code,
217 message=error_msg,
218 )
219 return Err(
220 WhatsAppNotificationError(error_msg, twilio_code=twilio_code)
221 )
223 async def _health_twilio(self, timeout: float) -> HealthCheckResult:
224 """Health check via Twilio Account API."""
225 import aiohttp
227 try:
228 url = f"{TWILIO_API_BASE}/Accounts/{self._account_sid}.json"
229 async with aiohttp.ClientSession() as session:
230 async with session.get(
231 url,
232 headers={"Authorization": self._twilio_auth_header()},
233 timeout=aiohttp.ClientTimeout(total=timeout),
234 ) as resp:
235 status = (
236 HealthStatus.HEALTHY
237 if resp.status < 500
238 else HealthStatus.UNHEALTHY
239 )
240 return HealthCheckResult(
241 component="whatsapp_twilio",
242 status=status,
243 details={"http_status": resp.status},
244 )
245 except OSError as exc:
246 return HealthCheckResult(
247 component="whatsapp_twilio",
248 status=HealthStatus.UNHEALTHY,
249 details={"error": str(exc)},
250 )
252 # ──────────────────────────────────────────────────────────────
253 # Meta provider
254 # ──────────────────────────────────────────────────────────────
256 async def _send_meta(
257 self, message: SMSMessage
258 ) -> Result[MessageDeliveryReceipt, NotificationError]:
259 """Send via Meta WhatsApp Business API."""
260 try:
261 import httpx
262 except ImportError as exc:
263 raise ImportError(
264 "httpx is required for the Meta WhatsApp provider. "
265 "Install with: pip install lexigram-notification[whatsapp]"
266 ) from exc
268 url = f"{_META_API_BASE}/{self._phone_number_id}/messages"
269 headers = {
270 "Authorization": f"Bearer {self._access_token}",
271 "Content-Type": "application/json",
272 }
273 to_number = message.to[0] if message.to else ""
274 payload: dict[str, Any] = {
275 "messaging_product": "whatsapp",
276 "to": to_number,
277 "type": "text",
278 "text": {"body": message.body},
279 }
281 async with httpx.AsyncClient(timeout=self._timeout) as client:
282 resp = await client.post(url, json=payload, headers=headers)
283 body = resp.json()
284 if resp.status_code == 200:
285 messages = body.get("messages", [{}])
286 wa_id = messages[0].get("id") if messages else None
287 receipt = MessageDeliveryReceipt(
288 message_id=str(uuid.uuid4()),
289 backend="whatsapp_meta",
290 channel="whatsapp",
291 provider_reference=wa_id,
292 )
293 logger.info(
294 "whatsapp.meta_sent",
295 to=message.to,
296 wa_id=wa_id,
297 )
298 return Ok(receipt)
300 error = body.get("error", {})
301 error_msg = error.get("message", f"HTTP {resp.status_code}")
302 meta_code = str(error.get("code", "")) or None
303 logger.warning(
304 "whatsapp.meta_send_failed",
305 status=resp.status_code,
306 error=error_msg,
307 code=meta_code,
308 )
309 return Err(WhatsAppMetaNotificationError(error_msg, meta_code=meta_code))
311 async def _health_meta(self, timeout: float) -> HealthCheckResult:
312 """Health check via Meta Graph API token debug endpoint."""
313 try:
314 import httpx
315 except ImportError:
316 return HealthCheckResult(
317 component="whatsapp_meta",
318 status=HealthStatus.UNHEALTHY,
319 message="httpx is not installed",
320 )
322 try:
323 url = (
324 f"https://graph.facebook.com/debug_token"
325 f"?input_token={self._access_token}"
326 f"&access_token={self._access_token}"
327 )
328 async with httpx.AsyncClient(timeout=timeout) as client:
329 resp = await client.get(url)
330 status = (
331 HealthStatus.HEALTHY
332 if resp.status_code < 500
333 else HealthStatus.UNHEALTHY
334 )
335 return HealthCheckResult(
336 component="whatsapp_meta",
337 status=status,
338 details={"http_status": resp.status_code},
339 )
340 except OSError as exc:
341 return HealthCheckResult(
342 component="whatsapp_meta",
343 status=HealthStatus.UNHEALTHY,
344 details={"error": str(exc)},
345 )
348__all__ = [
349 "WhatsAppBackend",
350 "WhatsAppMetaNotificationError",
351 "WhatsAppNotificationError",
352]