Coverage for src/lexigram/notification/mailer/sendgrid_mailer.py: 26%

1"""SendGrid email backend.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any 

6import uuid 

7 

8from lexigram.contracts.core import HealthCheckResult, HealthStatus 

9from lexigram.contracts.mailer import ( 

10 EmailMessage, 

11 MessageDeliveryReceipt, 

12) 

13from lexigram.logging import get_logger 

14from lexigram.notification.constants import DEFAULT_FROM_EMAIL, SENDGRID_API_URL 

15from lexigram.notification.exceptions import SendGridMailerError 

16from lexigram.result import Err, Ok, Result 

17 

18if TYPE_CHECKING: 

19 from lexigram.contracts.mailer.errors import MailerError 

20 

21logger = get_logger(__name__) 

22 

23 

24class SendGridMailer: 

25 """Email backend that sends via SendGrid REST API v3. 

26 

27 Implements :class:`~lexigram.contracts.mailer.protocols.MailerProtocol`. 

28 Requires the ``aiohttp`` optional extra. 

29 

30 Args: 

31 api_key: SendGrid API key (``SG.*``). 

32 timeout: HTTP request timeout in seconds. 

33 sandbox_mode: When ``True``, emails are not actually delivered. 

34 from_email: Default sender email address. 

35 from_name: Default sender display name. 

36 """ 

37 

38 def __init__( 

39 self, 

40 api_key: str, 

41 timeout: int = 30, 

42 sandbox_mode: bool = False, 

43 from_email: str | None = None, 

44 from_name: str | None = None, 

45 ) -> None: 

46 self._api_key = api_key 

47 self._timeout = timeout 

48 self._sandbox_mode = sandbox_mode 

49 self.from_email = from_email or DEFAULT_FROM_EMAIL 

50 self.from_name = from_name 

51 

52 def _build_payload(self, message: EmailMessage) -> dict[str, Any]: 

53 """Build SendGrid Mail Send API v3 JSON payload.""" 

54 sender_email = message.from_email or self.from_email 

55 sender_name = message.from_name or self.from_name 

56 from_obj: dict[str, Any] = {"email": sender_email} 

57 if sender_name: 

58 from_obj["name"] = sender_name 

59 

60 content = [] 

61 if message.body: 

62 content.append({"type": "text/plain", "value": message.body}) 

63 if message.html_body: 

64 content.append({"type": "text/html", "value": message.html_body}) 

65 

66 payload: dict[str, Any] = { 

67 "personalizations": [{"to": [{"email": r} for r in message.to]}], 

68 "from": from_obj, 

69 "subject": message.subject, 

70 "content": content, 

71 } 

72 

73 if message.cc: 

74 payload["personalizations"][0]["cc"] = [{"email": r} for r in message.cc] 

75 if message.bcc: 

76 payload["personalizations"][0]["bcc"] = [{"email": r} for r in message.bcc] 

77 if self._sandbox_mode: 

78 payload["mail_settings"] = {"sandbox_mode": {"enable": True}} 

79 

80 return payload 

81 

82 async def send( 

83 self, message: EmailMessage 

84 ) -> Result[MessageDeliveryReceipt, MailerError]: 

85 """Send an email via the SendGrid REST API. 

86 

87 Args: 

88 message: The email to deliver. 

89 

90 Returns: 

91 ``Ok(MessageDeliveryReceipt)`` on HTTP 202. 

92 ``Err(SendGridMailerError)`` for 4xx/5xx responses. 

93 

94 Raises: 

95 aiohttp.ClientError: For network connectivity issues (DNS, connection refused). 

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

97 

98 Note: 

99 Infrastructure errors propagate to enable retry and circuit-breaker logic 

100 at higher levels. Only HTTP-level errors are wrapped in the Result type. 

101 """ 

102 import aiohttp 

103 

104 payload = self._build_payload(message) 

105 headers = { 

106 "Authorization": f"Bearer {self._api_key}", 

107 "Content-Type": "application/json", 

108 } 

109 

110 async with aiohttp.ClientSession() as session: 

111 async with session.post( 

112 SENDGRID_API_URL, 

113 json=payload, 

114 headers=headers, 

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

116 ) as resp: 

117 if resp.status == 202: 

118 provider_ref = resp.headers.get("X-Message-Id") 

119 receipt = MessageDeliveryReceipt( 

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

121 backend="sendgrid", 

122 channel="email", 

123 provider_reference=provider_ref, 

124 ) 

125 logger.info("sendgrid_sent", to=message.to, msg_id=provider_ref) 

126 return Ok(receipt) 

127 

128 body = await resp.json() 

129 errors = body.get("errors", []) 

130 error_msg = errors[0]["message"] if errors else f"HTTP {resp.status}" 

131 logger.warning("sendgrid_send_failed", status=resp.status, body=body) 

132 return Err(SendGridMailerError(error_msg, status_code=resp.status)) 

133 

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

135 """Check SendGrid API reachability. 

136 

137 Args: 

138 timeout: Max seconds to wait. 

139 

140 Returns: 

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

142 """ 

143 import aiohttp 

144 

145 try: 

146 async with aiohttp.ClientSession() as session: 

147 async with session.head( 

148 SENDGRID_API_URL, 

149 headers={"Authorization": f"Bearer {self._api_key}"}, 

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

151 ) as resp: 

152 status = ( 

153 HealthStatus.HEALTHY 

154 if resp.status < 500 

155 else HealthStatus.UNHEALTHY 

156 ) 

157 return HealthCheckResult( 

158 component="sendgrid", 

159 status=status, 

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

161 ) 

162 except OSError as exc: 

163 return HealthCheckResult( 

164 component="sendgrid", 

165 status=HealthStatus.UNHEALTHY, 

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

167 ) 

168 

169 

170__all__ = ["SendGridMailer"]