Coverage for src/lexigram/notification/backends/sms/twilio.py: 100%

1"""Twilio SMS backend.""" 

2 

3from __future__ import annotations 

4 

5import base64 

6from typing import TYPE_CHECKING 

7import uuid 

8 

9from lexigram.contracts.core import HealthCheckResult, HealthStatus 

10from lexigram.contracts.mailer import MessageDeliveryReceipt 

11from lexigram.contracts.notification.types import SMSMessage 

12from lexigram.logging import get_logger 

13from lexigram.notification.constants import DEFAULT_TWILIO_TIMEOUT, TWILIO_API_BASE 

14from lexigram.notification.exceptions import TwilioNotificationError 

15from lexigram.result import Err, Ok, Result 

16 

17if TYPE_CHECKING: 

18 from lexigram.contracts.notification.errors import NotificationError 

19 

20logger = get_logger(__name__) 

21 

22 

23class TwilioSMS: 

24 """SMS backend that sends via Twilio REST API. 

25 

26 Implements :class:`~lexigram.contracts.notification.protocols.SMSChannelProtocol`. 

27 Requires the ``aiohttp`` optional extra. 

28 

29 Args: 

30 account_sid: Twilio Account SID. 

31 auth_token: Twilio Auth Token. 

32 from_number: Twilio phone number in E.164 format (e.g., +15550000000). 

33 timeout: HTTP request timeout in seconds. 

34 """ 

35 

36 def __init__( 

37 self, 

38 account_sid: str, 

39 auth_token: str, 

40 from_number: str | None = None, 

41 timeout: int = DEFAULT_TWILIO_TIMEOUT, 

42 ) -> None: 

43 self._account_sid = account_sid 

44 self._auth_token = auth_token 

45 self._from_number = from_number 

46 self._timeout = timeout 

47 

48 def _get_auth_header(self) -> str: 

49 """Generate Basic Auth header for Twilio.""" 

50 credentials = f"{self._account_sid}:{self._auth_token}" 

51 encoded = base64.b64encode(credentials.encode()).decode() 

52 return f"Basic {encoded}" 

53 

54 async def send( 

55 self, message: SMSMessage 

56 ) -> Result[MessageDeliveryReceipt, NotificationError]: 

57 """Send an SMS via the Twilio REST API. 

58 

59 Args: 

60 message: The SMS to deliver. 

61 

62 Returns: 

63 ``Ok(MessageDeliveryReceipt)`` on HTTP 201. 

64 ``Err(TwilioNotificationError)`` for 4xx/5xx responses. 

65 

66 Raises: 

67 aiohttp.ClientError: For network connectivity issues. 

68 OSError: For low-level socket/infrastructure errors. 

69 """ 

70 import aiohttp 

71 

72 url = f"{TWILIO_API_BASE}/Accounts/{self._account_sid}/Messages.json" 

73 headers = { 

74 "Authorization": self._get_auth_header(), 

75 "Content-Type": "application/x-www-form-urlencoded", 

76 } 

77 

78 # Build form data 

79 from_number = message.from_number or self._from_number 

80 data = { 

81 "To": message.to[0] if message.to else "", 

82 "From": from_number, 

83 "Body": message.body, 

84 } 

85 

86 async with aiohttp.ClientSession() as session: 

87 async with session.post( 

88 url, 

89 data=data, 

90 headers=headers, 

91 timeout=aiohttp.ClientTimeout(total=self._timeout), 

92 ) as resp: 

93 body = await resp.json() 

94 

95 if resp.status == 201: 

96 receipt = MessageDeliveryReceipt( 

97 message_id=str(uuid.uuid4()), 

98 backend="twilio", 

99 channel="sms", 

100 provider_reference=body.get("sid"), 

101 ) 

102 logger.info("twilio_sent", to=message.to, sid=body.get("sid")) 

103 return Ok(receipt) 

104 

105 # Handle error response 

106 error_msg = body.get("message", f"HTTP {resp.status}") 

107 twilio_code = body.get("code") 

108 logger.warning( 

109 "twilio_send_failed", 

110 status=resp.status, 

111 code=twilio_code, 

112 message=error_msg, 

113 ) 

114 return Err(TwilioNotificationError(error_msg, twilio_code=twilio_code)) 

115 

116 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult: 

117 """Check Twilio API reachability. 

118 

119 Args: 

120 timeout: Max seconds to wait. 

121 

122 Returns: 

123 :class:`~lexigram.contracts.core.HealthCheckResult`. 

124 """ 

125 import aiohttp 

126 

127 try: 

128 url = f"{TWILIO_API_BASE}/Accounts/{self._account_sid}.json" 

129 async with aiohttp.ClientSession() as session: 

130 async with session.get( 

131 url, 

132 headers={"Authorization": self._get_auth_header()}, 

133 timeout=aiohttp.ClientTimeout(total=timeout), 

134 ) as resp: 

135 status = ( 

136 HealthStatus.HEALTHY 

137 if resp.status < 500 

138 else HealthStatus.UNHEALTHY 

139 ) 

140 return HealthCheckResult( 

141 component="twilio", 

142 status=status, 

143 details={"http_status": resp.status}, 

144 ) 

145 except OSError as exc: 

146 return HealthCheckResult( 

147 component="twilio", 

148 status=HealthStatus.UNHEALTHY, 

149 details={"error": str(exc)}, 

150 ) 

151 

152 

153__all__ = ["TwilioSMS"]