import aiosmtplib from email.message import EmailMessage from email.mime.image import MIMEImage from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.policy import SMTP from src.application.contracts.i_sender import ISender class EmailSender(ISender): def __init__( self, host: str, port: int, username: str, password: str, from_addr: str, use_tls: bool = True, timeout: int = 10, ): self._host = host self._port = port self._username = username self._password = password self._from_addr = from_addr self._use_tls = use_tls self._timeout = timeout async def send( self, to: str, subject: str, body: str, plain: str | None = None, *, inline_png: tuple[str, bytes] | None = None, ) -> None: if inline_png and plain: message = self._multipart_with_inline_png( to=to, subject=subject, plain=plain, html=body, content_id=inline_png[0], png_bytes=inline_png[1], ) else: message = EmailMessage() message['From'] = self._from_addr message['To'] = to message['Subject'] = subject if plain: message.set_content(plain) message.add_alternative(body, subtype='html') else: message.set_content(body, subtype='html') await aiosmtplib.send( message, hostname=self._host, port=self._port, username=self._username, password=self._password, use_tls=self._use_tls, timeout=self._timeout, ) def _multipart_with_inline_png( self, *, to: str, subject: str, plain: str, html: str, content_id: str, png_bytes: bytes, ) -> MIMEMultipart: root = MIMEMultipart('alternative', policy=SMTP) root['Subject'] = subject root['From'] = self._from_addr root['To'] = to root.attach(MIMEText(plain, 'plain', 'utf-8')) related = MIMEMultipart('related', policy=SMTP) related.attach(MIMEText(html, 'html', 'utf-8')) image = MIMEImage(png_bytes, _subtype='png', policy=SMTP) image.add_header('Content-ID', f'<{content_id}>') image.add_header('Content-Disposition', 'inline', filename='exa-logo.png') related.attach(image) root.attach(related) return root