Раис Юсупалиев
137 lines
3.5 KiB
Python
137 lines
3.5 KiB
Python
"""SMTP e-mail adapter built on top of aiosmtplib."""
|
|
|
|
from email.message import EmailMessage
|
|
from typing import Protocol
|
|
|
|
import aiosmtplib
|
|
import structlog
|
|
|
|
log = structlog.get_logger(__name__)
|
|
|
|
|
|
class SMTPEmailSenderError(Exception):
|
|
"""Raised when the SMTP delivery fails."""
|
|
|
|
|
|
class SMTPSendFn(Protocol):
|
|
async def __call__(
|
|
self,
|
|
message: EmailMessage,
|
|
*,
|
|
hostname: str,
|
|
port: int,
|
|
username: str | None,
|
|
password: str | None,
|
|
use_tls: bool,
|
|
start_tls: bool,
|
|
timeout: float,
|
|
) -> object: ...
|
|
|
|
|
|
class SMTPEmailSender:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
smtp_host: str,
|
|
smtp_port: int,
|
|
username: str,
|
|
password: str,
|
|
from_address: str,
|
|
use_tls: bool = True,
|
|
timeout_seconds: float = 10.0,
|
|
send: SMTPSendFn | None = None,
|
|
) -> None:
|
|
self._smtp_host = smtp_host
|
|
self._smtp_port = smtp_port
|
|
self._username = username
|
|
self._password = password
|
|
self._from_address = from_address
|
|
self._use_tls = use_tls
|
|
self._timeout_seconds = timeout_seconds
|
|
self._send = send or _default_send
|
|
|
|
async def send_email(
|
|
self,
|
|
*,
|
|
to: str,
|
|
subject: str,
|
|
body: str,
|
|
attachment_bytes: bytes | None = None,
|
|
attachment_filename: str | None = None,
|
|
) -> None:
|
|
message = self._build_message(
|
|
to=to,
|
|
subject=subject,
|
|
body=body,
|
|
attachment_bytes=attachment_bytes,
|
|
attachment_filename=attachment_filename,
|
|
)
|
|
try:
|
|
await self._send(
|
|
message,
|
|
hostname=self._smtp_host,
|
|
port=self._smtp_port,
|
|
username=self._username or None,
|
|
password=self._password or None,
|
|
use_tls=self._use_tls and self._smtp_port == 465,
|
|
start_tls=self._use_tls and self._smtp_port != 465,
|
|
timeout=self._timeout_seconds,
|
|
)
|
|
except Exception as exc:
|
|
log.warning(
|
|
"smtp_send_failed",
|
|
smtp_host=self._smtp_host,
|
|
smtp_port=self._smtp_port,
|
|
to=to,
|
|
error=str(exc),
|
|
)
|
|
raise SMTPEmailSenderError(
|
|
f"SMTP send to {to} failed: {exc}"
|
|
) from exc
|
|
|
|
def _build_message(
|
|
self,
|
|
*,
|
|
to: str,
|
|
subject: str,
|
|
body: str,
|
|
attachment_bytes: bytes | None = None,
|
|
attachment_filename: str | None = None,
|
|
) -> EmailMessage:
|
|
message = EmailMessage()
|
|
message["From"] = self._from_address
|
|
message["To"] = to
|
|
message["Subject"] = subject
|
|
message.set_content(body)
|
|
if attachment_bytes is not None and attachment_filename is not None:
|
|
message.add_attachment(
|
|
attachment_bytes,
|
|
maintype="application",
|
|
subtype="pdf",
|
|
filename=attachment_filename,
|
|
)
|
|
return message
|
|
|
|
|
|
async def _default_send(
|
|
message: EmailMessage,
|
|
*,
|
|
hostname: str,
|
|
port: int,
|
|
username: str | None,
|
|
password: str | None,
|
|
use_tls: bool,
|
|
start_tls: bool,
|
|
timeout: float,
|
|
) -> object:
|
|
return await aiosmtplib.send(
|
|
message,
|
|
hostname=hostname,
|
|
port=port,
|
|
username=username,
|
|
password=password,
|
|
use_tls=use_tls,
|
|
start_tls=start_tls,
|
|
timeout=timeout,
|
|
)
|