from __future__ import annotations from dataclasses import dataclass from email.message import EmailMessage from pathlib import Path import aiosmtplib @dataclass class MailSendResult: ok: bool error: str | None class MailNotifier: def __init__( self, smtp_host: str, smtp_port: int, username: str, password: str, from_addr: str, to_addr: str, use_starttls: bool = True, use_tls: bool = False, timeout_seconds: float = 15, ) -> None: self.smtp_host = smtp_host self.smtp_port = smtp_port self.username = username self.password = password self.from_addr = from_addr self.to_addr = to_addr self.use_starttls = use_starttls self.use_tls = use_tls self.timeout_seconds = timeout_seconds async def send_message( self, subject: str, body: str, attachments: list[str] | None = None, ) -> MailSendResult: message = EmailMessage() message["From"] = self.from_addr message["To"] = self.to_addr message["Subject"] = subject message.set_content(body) for attachment in attachments or []: path = Path(attachment) if not path.exists() or not path.is_file(): continue data = path.read_bytes() message.add_attachment( data, maintype="image", subtype="png", filename=path.name, ) try: await aiosmtplib.send( message, hostname=self.smtp_host, port=self.smtp_port, username=self.username, password=self.password, start_tls=self.use_starttls, use_tls=self.use_tls, timeout=self.timeout_seconds, ) return MailSendResult(ok=True, error=None) except Exception as exc: return MailSendResult(ok=False, error=str(exc))