DataOps-platform/app/core/events/email_delivery.py

"""Small SMTP binding for work-center email delivery."""

from __future__ import annotations

import os
import smtplib
from email.message import EmailMessage


def smtp_sender(notification: dict) -> None:
    """Deliver one notification; configuration failures remain retryable evidence."""
    host = os.getenv("DATAOPS_SMTP_HOST")
    sender = os.getenv("DATAOPS_SMTP_FROM")
    recipient = notification.get("recipient_email")
    if not host or not sender or not recipient:
        raise RuntimeError("SMTP binding is incomplete")
    port = int(os.getenv("DATAOPS_SMTP_PORT", "587"))
    message = EmailMessage()
    message["From"] = sender
    message["To"] = recipient
    message["Subject"] = notification["subject"]
    message.set_content(notification["body"])
    with smtplib.SMTP(host, port, timeout=10) as client:
        if os.getenv("DATAOPS_SMTP_STARTTLS", "true").lower() == "true":
            client.starttls()
        username = os.getenv("DATAOPS_SMTP_USERNAME")
        password = os.getenv("DATAOPS_SMTP_PASSWORD")
        if username and password:
            client.login(username, password)
        client.send_message(message)