Coverage for apps/communications/delivery.py: 95%
66 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-04 14:06 +0000
1"""Outbound email delivery through Twilio SendGrid (add-sendgrid-email).
3Same port-and-stub pattern as the other external seams (``underwriting/providers.py``):
5- ``EmailSender`` is the port; ``SendGridEmailSender`` is the real adapter and
6 ``SimulatedEmailSender`` is used when no SendGrid key is configured (local dev, tests).
7- ``deliver_email`` is the one place every email goes through: it records the ``Communication``
8 (provider, provider message id, status) and its first event, so Staff LOS → Communications shows
9 every email and the SendGrid webhook can later move it to Delivered / Opened.
10- Every email is sent as plain text plus the branded HTML layout (``email_content.py``,
11 brand-email-template).
12"""
14from __future__ import annotations
16import logging
17from dataclasses import dataclass
18from typing import Protocol
20from django.conf import settings
22from apps.communications.email_content import EmailContent, render_html
23from apps.communications.models import Communication, CommunicationEvent
25logger = logging.getLogger(__name__)
28class EmailSendError(Exception):
29 """The provider refused or failed to accept the email."""
32@dataclass(frozen=True)
33class SendResult:
34 provider: str
35 message_id: str | None
38class EmailSender(Protocol):
39 provider: str
41 def send(
42 self, *, to: str, subject: str, body: str, html: str, comm_id: int, track_clicks: bool
43 ) -> SendResult: ...
46class SimulatedEmailSender:
47 """No SendGrid key: nothing leaves the server; the message is only recorded."""
49 provider = "simulated"
51 def send(self, *, to, subject, body, html, comm_id, track_clicks) -> SendResult:
52 return SendResult(provider=self.provider, message_id=None)
55class SendGridEmailSender:
56 provider = "sendgrid"
58 def __init__(self, api_key: str, from_email: str, from_name: str):
59 self.api_key = api_key
60 self.from_email = from_email
61 self.from_name = from_name
63 def send(self, *, to, subject, body, html, comm_id, track_clicks) -> SendResult:
64 from sendgrid import SendGridAPIClient
65 from sendgrid.helpers.mail import (
66 ClickTracking,
67 CustomArg,
68 Email,
69 Mail,
70 OpenTracking,
71 TrackingSettings,
72 )
74 message = Mail(
75 from_email=Email(self.from_email, self.from_name),
76 to_emails=to,
77 subject=subject,
78 plain_text_content=body,
79 html_content=html,
80 )
81 # Open tracking feeds "Opened"; links are only rewritten for click tracking where it is
82 # safe (never for password / reset emails).
83 message.tracking_settings = TrackingSettings(
84 click_tracking=ClickTracking(track_clicks, track_clicks),
85 open_tracking=OpenTracking(True),
86 )
87 # Every webhook event carries this back, so it maps to our Communication row.
88 message.custom_arg = CustomArg("comm_id", str(comm_id))
89 try:
90 response = SendGridAPIClient(self.api_key).send(message)
91 except Exception as exc: # network error or 4xx/5xx raised by the SDK
92 raise EmailSendError(str(exc)) from exc
93 if response.status_code not in (200, 202): 93 ↛ 94line 93 didn't jump to line 94 because the condition on line 93 was never true
94 raise EmailSendError(f"SendGrid answered {response.status_code}")
95 headers = getattr(response, "headers", {}) or {}
96 message_id = headers.get("X-Message-Id") or headers.get("x-message-id")
97 return SendResult(provider=self.provider, message_id=message_id)
100def get_email_sender() -> EmailSender:
101 """SendGrid when a key and a from-address are configured, otherwise simulated."""
102 if settings.SENDGRID_API_KEY and settings.SENDGRID_FROM_EMAIL:
103 return SendGridEmailSender(
104 settings.SENDGRID_API_KEY, settings.SENDGRID_FROM_EMAIL, settings.SENDGRID_FROM_NAME
105 )
106 return SimulatedEmailSender()
109def deliver_email(
110 *,
111 to: str,
112 subject: str,
113 body: str | None = None,
114 content: EmailContent | None = None,
115 trigger: str | None = None,
116 loan=None,
117 automated: bool = False,
118 template=None,
119 track_clicks: bool = True,
120 raise_on_failure: bool = False,
121) -> Communication:
122 """Send one email and record it.
124 Pass ``content`` (system emails, see ``emails.py``) or plain ``body`` text (staff Compose);
125 either way the email goes out as plain text plus the branded HTML layout, and the plain text is
126 what is stored on the ``Communication``.
128 The ``Communication`` is created first (its id goes to SendGrid as ``comm_id``), then sent.
129 Success → status Sent + a ``sent`` event (+ the provider message id). Failure → status Failed
130 + a ``failed`` event; the error is logged, never shown to users. With ``raise_on_failure`` the
131 caller gets ``EmailSendError`` (e.g. an invite that must not be created without its email).
132 """
133 if content is None:
134 content = EmailContent.from_text(body or "")
135 body = content.to_text()
136 html = render_html(content, subject)
137 sender = get_email_sender()
138 comm = Communication.objects.create(
139 loan=loan,
140 channel="email",
141 template=template,
142 to=to,
143 subject=subject,
144 body=body,
145 status="Queued",
146 provider=sender.provider,
147 automated=automated,
148 trigger=trigger,
149 )
150 try:
151 result = sender.send(
152 to=to, subject=subject, body=body, html=html, comm_id=comm.id,
153 track_clicks=track_clicks,
154 )
155 except EmailSendError as exc:
156 logger.warning("Email %s (%s) to %s failed: %s", comm.id, trigger, to, exc)
157 comm.status = "Failed"
158 comm.save(update_fields=["status", "updated_at"])
159 CommunicationEvent.objects.create(communication=comm, event_type="failed")
160 if raise_on_failure:
161 raise
162 return comm
163 comm.provider_message_id = result.message_id
164 comm.status = "Sent"
165 comm.save(update_fields=["provider_message_id", "status", "updated_at"])
166 CommunicationEvent.objects.create(communication=comm, event_type="sent")
167 return comm