""" A class for composing and sending email messages with support for text, HTML, attachments, and multiple recipients. """ import base64 from email import encoders from email.mime.base import MIMEBase from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from typing import Optional from pydantic import EmailStr from common.src.enums import MimeSubTypes from common.src.typings import Base64String from ..constants import EmailPayload, EmailHeaders from .models import BaseAttachment class EmailComposer: """Class for building and sending email messages.""" def __init__( self, source: str | EmailStr, to: list[str | EmailStr], subject: str, text: str, html: Optional[str] = None, cc: Optional[list[str | EmailStr]] = None, bcc: Optional[list[str | EmailStr]] = None, reply_to: Optional[list[str | EmailStr]] = None, attachments: Optional[list[BaseAttachment]] = None, ): """ Represents an email message with specified recipients, content, and optional additional fields such as CC, BCC, and attachments. Args: source: Required sender email address. to: Required list of recipient email addresses. subject: Required subject of the email. text: Required plain text content of the email. html: Optional HTML content of the email. to: Optional list of CC email addresses. cc: Optional list of BCC email addresses. bcc: Optional list of reply-to email addresses. attachments: Optional list of attachment objects. """ self._subject = subject self._text = text self._html = html self._from = source self._to = to self._cc = cc or [] self._bcc = bcc or [] self._reply_to = reply_to or [] self._attachments = attachments or [] @property def recipients(self) -> list[str]: """Get all unique recipients, including CC and BCC.""" return list(set(self._to + self._cc + self._bcc)) def compose(self) -> MIMEMultipart: """Compose the email message. This method builds the MIME message, attaches the body, and any provided attachments, outputting the final message as an object, ready to be sent. """ msg = self._build_mime() self._attach_body(msg) for attachment in self._attachments: self._attach_attachment(msg, attachment) return msg def _build_mime(self) -> MIMEMultipart: """Build the MIME message.""" msg = MIMEMultipart("mixed") msg[EmailPayload.SUBJECT] = self._subject msg[EmailPayload.FROM] = self._from msg[EmailPayload.TO] = self._concatenate(self._to) if self._cc: msg[EmailPayload.CC] = self._concatenate(self._cc) if self._bcc: msg[EmailPayload.BCC] = self._concatenate(self._bcc) if self._reply_to: msg.add_header(EmailHeaders.REPLY_TO, self._concatenate(self._reply_to)) return msg def _attach_body(self, msg: MIMEMultipart) -> None: """Attach the body of the email. Args: msg: The MIME message object to attach the body to. """ msg_body = MIMEMultipart(MimeSubTypes.ALTERNATIVE) msg_body.attach(MIMEText(self._text, MimeSubTypes.PLAIN)) if self._html: msg_body.attach(MIMEText(self._html, MimeSubTypes.HTML)) msg.attach(msg_body) @staticmethod def _attach_attachment(msg: MIMEMultipart, attachment: BaseAttachment) -> None: """Attach any attachments to the email. Args: msg: The MIME message object to attach the attachments to. attachment: The attachment object to attach. """ mime_type, mime_subtype = attachment.mime_type.split("/", 1) mime_attachment = MIMEBase(mime_type, mime_subtype) # If necessary, convert the attachment data bytes into as a base64 # encoded string and attach it to the message. b64_payload: Base64String = ( attachment.attachment if attachment.is_b64_string else base64.b64encode(attachment.attachment_data).decode() ) mime_attachment.set_payload(b64_payload) encoders.encode_base64(mime_attachment) mime_attachment.add_header( EmailHeaders.CONTENT_DISPOSITION, f'attachment; filename="{attachment.file_name}"', ) msg.attach(mime_attachment) @staticmethod def _concatenate(emails: list[str | EmailStr]) -> str: """Concatenate emails into a single string.""" return ",".join(emails)