mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-19 17:03:02 +00:00
71 lines
2.2 KiB
Python
71 lines
2.2 KiB
Python
"""SES-over-SMTP implementation of the outbound-email port (ADR-0059).
|
|
|
|
Sends via `smtplib` against the SES SMTP endpoint using the credentials
|
|
provisioned in terraform (`modules/ses`, Secrets Manager
|
|
`${stage}/ses/smtp_credentials`). The SMTP transport is injected so the sender
|
|
is testable without a live server.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import smtplib
|
|
from email.message import EmailMessage
|
|
from types import TracebackType
|
|
from typing import Any, Callable, Optional, Protocol, Type, cast
|
|
|
|
|
|
class SmtpTransport(Protocol):
|
|
"""The minimal SMTP surface the sender uses — satisfied structurally by
|
|
`smtplib.SMTP` and by test fakes."""
|
|
|
|
def __enter__(self) -> "SmtpTransport": ...
|
|
|
|
def __exit__(
|
|
self,
|
|
exc_type: Optional[Type[BaseException]],
|
|
exc: Optional[BaseException],
|
|
tb: Optional[TracebackType],
|
|
) -> Optional[bool]: ...
|
|
|
|
def starttls(self) -> Any: ...
|
|
|
|
def login(self, user: str, password: str) -> Any: ...
|
|
|
|
def send_message(self, message: EmailMessage) -> Any: ...
|
|
|
|
|
|
def _default_smtp(host: str, port: int) -> SmtpTransport:
|
|
# `smtplib.SMTP` satisfies SmtpTransport structurally; its stdlib stubs
|
|
# declare wider signatures (extra optional args, an SMTP-typed context
|
|
# manager), so the cast bridges the real transport to the port's surface.
|
|
return cast(SmtpTransport, smtplib.SMTP(host, port))
|
|
|
|
|
|
class SesSmtpEmailSender:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
host: str,
|
|
port: int,
|
|
username: str,
|
|
password: str,
|
|
from_address: str,
|
|
smtp_factory: Callable[[str, int], SmtpTransport] = _default_smtp,
|
|
) -> None:
|
|
self._host = host
|
|
self._port = port
|
|
self._username = username
|
|
self._password = password
|
|
self._from_address = from_address
|
|
self._smtp_factory = smtp_factory
|
|
|
|
def send(self, *, to: str, subject: str, body: str) -> None:
|
|
message = EmailMessage()
|
|
message["From"] = self._from_address
|
|
message["To"] = to
|
|
message["Subject"] = subject
|
|
message.set_content(body)
|
|
with self._smtp_factory(self._host, self._port) as smtp:
|
|
smtp.starttls()
|
|
smtp.login(self._username, self._password)
|
|
smtp.send_message(message)
|