94 lines
2.5 KiB
Python
94 lines
2.5 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import smtplib
|
||
|
|
from email.message import EmailMessage
|
||
|
|
from typing import TYPE_CHECKING
|
||
|
|
|
||
|
|
from app.core.config import settings
|
||
|
|
|
||
|
|
if TYPE_CHECKING:
|
||
|
|
from collections.abc import Sequence
|
||
|
|
|
||
|
|
|
||
|
|
def send_email(
|
||
|
|
*,
|
||
|
|
subject: str,
|
||
|
|
recipients: Sequence[str],
|
||
|
|
text_body: str,
|
||
|
|
html_body: str | None = None,
|
||
|
|
reply_to: str | None = None,
|
||
|
|
) -> None:
|
||
|
|
host = settings.SMTP_HOST
|
||
|
|
from_email = settings.SMTP_FROM_EMAIL
|
||
|
|
recipient_list = list(recipients)
|
||
|
|
encryption = _get_smtp_encryption()
|
||
|
|
|
||
|
|
if not host:
|
||
|
|
raise ValueError("SMTP_HOST must be configured")
|
||
|
|
|
||
|
|
if not from_email:
|
||
|
|
raise ValueError("SMTP_FROM_EMAIL must be configured")
|
||
|
|
|
||
|
|
if not recipient_list:
|
||
|
|
raise ValueError("At least one recipient is required")
|
||
|
|
|
||
|
|
message = EmailMessage()
|
||
|
|
message["Subject"] = subject
|
||
|
|
message["From"] = _format_sender(from_email=from_email, from_name=settings.SMTP_FROM_NAME)
|
||
|
|
message["To"] = ", ".join(recipient_list)
|
||
|
|
|
||
|
|
if reply_to:
|
||
|
|
message["Reply-To"] = reply_to
|
||
|
|
|
||
|
|
message.set_content(text_body)
|
||
|
|
|
||
|
|
if html_body is not None:
|
||
|
|
message.add_alternative(html_body, subtype="html")
|
||
|
|
|
||
|
|
smtp_client = _create_smtp_client(
|
||
|
|
host=host,
|
||
|
|
port=settings.SMTP_PORT,
|
||
|
|
encryption=encryption,
|
||
|
|
timeout=settings.SMTP_TIMEOUT,
|
||
|
|
)
|
||
|
|
|
||
|
|
try:
|
||
|
|
smtp_client.ehlo()
|
||
|
|
|
||
|
|
if encryption == "starttls":
|
||
|
|
smtp_client.starttls()
|
||
|
|
smtp_client.ehlo()
|
||
|
|
|
||
|
|
if settings.SMTP_USERNAME:
|
||
|
|
if settings.SMTP_PASSWORD is None:
|
||
|
|
raise ValueError("SMTP_PASSWORD must be configured when SMTP_USERNAME is set")
|
||
|
|
|
||
|
|
smtp_client.login(settings.SMTP_USERNAME, settings.SMTP_PASSWORD)
|
||
|
|
|
||
|
|
smtp_client.send_message(message)
|
||
|
|
finally:
|
||
|
|
smtp_client.quit()
|
||
|
|
|
||
|
|
|
||
|
|
def _format_sender(*, from_email: str, from_name: str | None) -> str:
|
||
|
|
if not from_name:
|
||
|
|
return from_email
|
||
|
|
|
||
|
|
return f"{from_name} <{from_email}>"
|
||
|
|
|
||
|
|
|
||
|
|
def _get_smtp_encryption() -> str:
|
||
|
|
encryption = settings.SMTP_ENCRYPTION.strip().lower()
|
||
|
|
if encryption in {"ssl/tls", "ssl-tls", "tls"}:
|
||
|
|
return "ssl_tls"
|
||
|
|
|
||
|
|
if encryption not in {"starttls", "ssl_tls"}:
|
||
|
|
raise ValueError("SMTP_ENCRYPTION must be 'starttls' or 'ssl_tls'")
|
||
|
|
|
||
|
|
return encryption
|
||
|
|
|
||
|
|
|
||
|
|
def _create_smtp_client(*, host: str, port: int, encryption: str, timeout: float) -> smtplib.SMTP:
|
||
|
|
client_class = smtplib.SMTP_SSL if encryption == "ssl_tls" else smtplib.SMTP
|
||
|
|
return client_class(host, port, timeout=timeout)
|