Coverage for functions \ flipdare \ mailer \ _mailer.py: 84%
197 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-05-08 12:22 +1000
1#!/usr/bin/env python
2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved.
3#
4# This file is part of Flipdare's proprietary software and contains
5# confidential and copyrighted material. Unauthorised copying,
6# modification, distribution, or use of this file is strictly
7# prohibited without prior written permission from Flipdare Pty Ltd.
8#
9# This software includes third-party components licensed under MIT,
10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details.
11#
13import email.mime.text as mime
14import smtplib
15import ssl
16from abc import ABC, abstractmethod
17from email.mime import multipart
18from email.mime.image import MIMEImage
19from typing import Any, NoReturn
20import minify_html as minifier
21from dataclasses import dataclass
22from flipdare.mailer.email_image import EmailImage
23from flipdare.app_log import LOG
24from flipdare.constants import INTERNAL_ADMIN_EMAIL, IS_DEBUG, IS_TRACE, SMTP_STATUS_OK
25from flipdare.core.singleton import Singleton
26from flipdare.mailer._jinja_email_template import JinjaEmailTemplate
27from flipdare.error.app_error import ServerError
28from flipdare.generated.shared.app_error_code import AppErrorCode
29from flipdare.util.mail_reply_type import MailReplyType
31__all__ = ["MailerOptions"]
34@dataclass
35class MailerOptions:
36 smtp_gateway: str
37 smtp_port: int
38 smtp_username: str
39 smtp_password: str
40 smtp_timeout: int
41 internal_service_email: str
42 def_mail_from: str
43 def_mail_reply: str
44 def_no_reply: str
46 @classmethod
47 def create(
48 cls,
49 smtp_gateway: str | None = None,
50 smtp_port: int | None = None,
51 smtp_username: str | None = None,
52 smtp_password: str | None = None,
53 smtp_timeout: int | None = None,
54 internal_service_email: str | None = None,
55 def_mail_from: str | None = None,
56 def_mail_reply: str | None = None,
57 def_no_reply: str | None = None,
58 ) -> "MailerOptions":
59 from flipdare.app_config import AppConfig, get_app_config
61 config: AppConfig = get_app_config()
62 return cls(
63 smtp_gateway=smtp_gateway or config.smtp_gateway,
64 smtp_port=smtp_port or config.smtp_port,
65 smtp_username=smtp_username or config.smtp_username,
66 smtp_password=smtp_password or config.smtp_password,
67 smtp_timeout=smtp_timeout or config.smtp_timeout,
68 internal_service_email=internal_service_email or INTERNAL_ADMIN_EMAIL,
69 def_mail_from=def_mail_from or MailReplyType.ADMIN.formatted,
70 def_mail_reply=def_mail_reply or MailReplyType.SUPPORT.formatted,
71 def_no_reply=def_no_reply or MailReplyType.NO_REPLY.formatted,
72 )
75class Mailer(ABC, Singleton):
76 def __init__(
77 self,
78 options: MailerOptions | None = None,
79 smtp_client: smtplib.SMTP_SSL | None = None,
80 ) -> None:
81 if options is None:
82 options = MailerOptions.create()
83 self._options = options
84 self._manager = ClientManager(options=options, smtp_client=smtp_client)
86 @abstractmethod
87 def send(self, *args: Any, **kwargs: Any) -> bool: ...
89 @property
90 def options(self) -> MailerOptions:
91 return self._options
93 @property
94 def smtp_client(self) -> smtplib.SMTP_SSL:
95 return self._manager.client
97 @staticmethod
98 def minify_html(html_content: str) -> str:
99 try:
100 # pylint: disable=no-member
101 return minifier.minify(
102 html_content,
103 # NOTE: default is false for noncompliant attributes
104 # NOTE: should keep that, unless we find a use case otherwise
105 # allow_noncompliant_unquoted_attribute_values=True
106 keep_html_and_head_opening_tags=True,
107 keep_closing_tags=True,
108 minify_js=True,
109 minify_css=True,
110 )
111 except Exception as e:
112 LOG().warning(f"HTML minification failed: {e}. Returning original HTML content.")
113 return html_content
115 def _send_template(
116 self,
117 email: str,
118 email_template: JinjaEmailTemplate[Any],
119 should_minify: bool,
120 no_reply: bool = True,
121 images: list[EmailImage] | None = None,
122 ) -> bool:
124 if IS_TRACE:
125 msg = f"Rendering email template {email_template.__class__.__name__} for recipient {email}"
126 LOG().trace(msg)
128 return self._send_raw(
129 email=email,
130 subject=email_template.subject,
131 html_str=email_template.html,
132 text_str=email_template.text,
133 should_minify=should_minify,
134 no_reply=no_reply,
135 images=images,
136 )
138 def _send_raw(
139 self,
140 email: str,
141 subject: str,
142 html_str: str,
143 text_str: str,
144 should_minify: bool,
145 no_reply: bool = True,
146 images: list[EmailImage] | None = None,
147 ) -> bool:
148 if IS_TRACE:
149 LOG().trace(f"Sending email to {email} with subject '{subject}'")
151 email_msg: multipart.MIMEMultipart | None = None
152 try:
153 email_msg = self._build(
154 email,
155 subject,
156 html_str,
157 text_str,
158 should_minify=should_minify,
159 no_reply=no_reply,
160 images=images,
161 )
162 except Exception as e:
163 msg = f"Error building email message for {email}: {e}"
164 self._raise_error(
165 message=msg,
166 error_code=AppErrorCode.EMAIL_TEMPLATE,
167 ex=e,
168 )
170 try:
171 smtp_client = self.smtp_client
172 smtp_client.send_message(email_msg)
173 if IS_DEBUG:
174 LOG().debug(f"Message successfully sent to {email}")
176 return True
177 except smtplib.SMTPException as smtp_ex:
178 msg = f"SMTP error sending email to {email}: {smtp_ex}"
179 self._raise_error(
180 message=msg,
181 error_code=AppErrorCode.EMAIL_SMTP,
182 ex=smtp_ex,
183 )
184 except Exception as ex:
185 msg = f"Error sending email to {email}: {ex}"
186 self._raise_error(message=msg, ex=ex)
188 def _raise_error(
189 self,
190 message: str,
191 ex: Exception,
192 error_code: AppErrorCode = AppErrorCode.EMAIL,
193 ) -> NoReturn:
194 LOG().error(f"{message}: {ex}")
195 raise ServerError(
196 message=message,
197 error_code=error_code,
198 ) from ex
200 def _build(
201 self,
202 email: str,
203 subject: str,
204 html_str: str,
205 text_str: str,
206 should_minify: bool,
207 no_reply: bool = True,
208 images: list[EmailImage] | None = None,
209 ) -> multipart.MIMEMultipart:
210 mail_from = self.options.def_mail_from
211 no_reply_email = self.options.def_no_reply
212 mail_reply_email = self.options.def_mail_reply
214 if IS_TRACE:
215 LOG().trace(
216 f"Building email message: From={mail_from}, To={email}, Subject='{subject}', "
217 f"Reply-To={no_reply_email if no_reply else mail_reply_email}, "
218 f"HTML Length={len(html_str)}, Text Length={len(text_str)}",
219 )
221 try:
222 actual_html = self.minify_html(html_str) if should_minify else html_str
223 reply_to = no_reply_email if no_reply else mail_reply_email
225 text_part = mime.MIMEText(text_str, "plain")
226 html_part = mime.MIMEText(actual_html, "html")
228 if not images:
229 if IS_TRACE:
230 LOG().trace(
231 "Email has no images; building simple multipart/alternative message"
232 )
234 msg = multipart.MIMEMultipart("alternative")
235 msg["Subject"] = subject
236 msg["From"] = mail_from
237 msg["Reply-To"] = reply_to
238 msg["To"] = email
239 msg.attach(text_part)
240 msg.attach(html_part)
241 else:
242 if IS_TRACE:
243 LOG().trace(f"Email has {len(images)} images to embed")
245 # multipart/related wraps HTML + inline CID images
246 related = multipart.MIMEMultipart("related")
247 related.attach(html_part)
248 for img in images:
249 mime_img = MIMEImage(img.buffer_bytes)
250 mime_img["Content-ID"] = f"<{img.cid}>"
251 mime_img.add_header("Content-Disposition", "inline", filename=f"{img.cid}.png")
252 related.attach(mime_img)
254 # multipart/alternative provides text/plain fallback
255 alternative = multipart.MIMEMultipart("alternative")
256 alternative.attach(text_part)
257 alternative.attach(related)
259 msg = multipart.MIMEMultipart("mixed")
260 msg["Subject"] = subject
261 msg["From"] = mail_from
262 msg["Reply-To"] = reply_to
263 msg["To"] = email
264 msg.attach(alternative)
266 return msg
268 except Exception as e:
269 error_msg = f"Error building email message for {email}: {e}"
270 self._raise_error(
271 message=error_msg,
272 error_code=AppErrorCode.EMAIL_TEMPLATE,
273 ex=e,
274 )
277class ClientManager:
278 __slots__ = (
279 "_client",
280 "password",
281 "smtp_gateway",
282 "smtp_port",
283 "smtp_timeout",
284 "username",
285 )
287 def __init__(
288 self,
289 options: MailerOptions,
290 smtp_client: smtplib.SMTP_SSL | None = None,
291 ) -> None:
292 self._client = smtp_client
293 self.smtp_gateway = options.smtp_gateway
294 self.smtp_port = options.smtp_port
295 self.smtp_timeout = options.smtp_timeout
296 self.username = options.smtp_username
297 self.password = options.smtp_password
299 @property
300 def client(self) -> smtplib.SMTP_SSL:
301 client = self._client
303 if client is None:
304 LOG().warning("No existing SMTP client; creating a new connection")
305 client = self._create()
306 else:
307 is_connected = self._is_connected()
308 if is_connected:
309 LOG().debug("SMTP server connection is active")
310 return client
311 else:
312 client = None # force re-creation
313 client = self._create()
314 msg = "SMTP server connection was closed; creating a new connection"
315 LOG().warning(msg)
317 self._client = client
318 return client
320 @client.setter
321 def client(self, client: smtplib.SMTP_SSL) -> None:
322 self._client = client
324 def _create(self) -> smtplib.SMTP_SSL:
325 """
326 Establishes a new connection to the SMTP server.
327 """
328 smtp_gateway = self.smtp_gateway
329 smtp_port = self.smtp_port
330 smtp_timeout = self.smtp_timeout
331 username, password = self._validate_credentials()
333 msg = f"Creating SMTP connection to {smtp_gateway}:{smtp_port} with timeout {smtp_timeout}"
334 LOG().info(msg)
336 try:
337 ctx = ssl.create_default_context()
338 conn = smtplib.SMTP_SSL(smtp_gateway, smtp_port, timeout=smtp_timeout, context=ctx)
339 conn.login(username, password)
341 if IS_TRACE:
342 LOG().trace(f"Starting TLS for SMTP connection to server {conn.local_hostname}")
344 return conn
345 except Exception as e:
346 msg = f"Error creating SMTP connection to {smtp_gateway}:{smtp_port} - {e}"
347 LOG().error(msg)
348 raise ServerError(message=msg, error_code=AppErrorCode.EMAIL_SMTP) from e
350 def _validate_credentials(self) -> tuple[str, str]:
351 username = self.username
352 password = self.password
353 if not username or not password:
354 msg = "SMTP username or password not set; cannot create SMTP connection"
355 raise ServerError(message=msg, error_code=AppErrorCode.EMAIL)
357 return username, password
359 def _is_connected(self) -> bool:
360 """
361 Checks if the SMTP connection is still open by sending a NOOP command.
362 Returns True if connected, False otherwise.
363 """
364 try:
365 client = self._client
366 if client is None:
367 return False
369 status = client.noop()[0]
370 if IS_TRACE:
371 LOG().trace(f"SMTP NOOP command returned status {status}")
373 except smtplib.SMTPServerDisconnected:
374 status = -1 # A non-250 status
375 LOG().warning("SMTP server disconnected during NOOP check")
376 except Exception as e:
377 LOG().error(f"An unexpected error occurred while checking connection: {e}")
378 status = -1
380 ok = status == SMTP_STATUS_OK
381 msg = f"SMTP connection check: status={status}, connected={ok}"
382 if IS_TRACE:
383 LOG().trace(msg)
385 return ok