Send email notifications

The pattern below uses Python's standard smtplib with Gmail SMTP, but the same shape works for any SMTP server (Outlook, Workspace SMTP relay, or transactional providers like Resend, SendGrid, Mailgun).

Prerequisites

Generate a Gmail App Password, a 16-character credential that lets SMTP authenticate without your real password:

1. Make sure 2-Step Verification is enabled on the Google Account.
2. Open https://myaccount.google.com/apppasswords.
3. Create a new password and name it, e.g. "dltHub pipeline".
4. Copy the 16 characters. Google displays them with spaces (abcd efgh ijkl mnop); the spaces are decorative, so strip them.

App Passwords don't affect normal sign-in: your password, 2FA, and existing sessions are unchanged. You can revoke the App Password from the same page without touching the account.

If you're on a Google Workspace domain, an administrator may have disabled App Passwords org-wide. In that case use a transactional provider (Resend, SendGrid, Mailgun). The wiring is the same; just point smtplib at their SMTP server and use their API key as the password.

Store credentials in your prod profile

Add to .dlt/prod.secrets.toml:

[notifications.email]
host = "smtp.gmail.com"
port = 587
sender = "you@example.com"
recipient = "you@example.com"
password = "abcdefghijklmnop"     # 16 chars, no spaces, no angle brackets

sender must be the same Google Account the App Password was generated on.

Allowlist outbound IPs

If your SMTP server requires IP allowlisting, enable static egress IPs so the job's outbound traffic uses a known set of source IPs.

Wire it into your pipeline

import smtplib
import time
from datetime import datetime, timezone
from email.message import EmailMessage

import dlt
from dlt.hub import run


def send_email(subject: str, body: str) -> None:
    host = dlt.secrets["notifications.email.host"]
    port = int(dlt.secrets["notifications.email.port"])
    sender = dlt.secrets["notifications.email.sender"]
    recipient = dlt.secrets["notifications.email.recipient"]
    password = dlt.secrets["notifications.email.password"]

    msg = EmailMessage()
    msg["Subject"] = subject
    msg["From"] = sender
    msg["To"] = recipient
    msg.set_content(body)

    with smtplib.SMTP(host, port) as s:
        s.starttls()
        s.login(sender, password)
        s.send_message(msg)


@run.pipeline("my_pipeline")
def my_job():
    pipeline = dlt.pipeline(
        pipeline_name="my_pipeline",
        destination="warehouse",
        dataset_name="my_dataset",
    )

    started = time.time()
    try:
        load_info = pipeline.run(my_source())
        send_email(
            f"[dltHub] {pipeline.pipeline_name} succeeded",
            "\n".join([
                f"Pipeline:  {pipeline.pipeline_name}",
                f"Status:    SUCCESS",
                f"Finished:  {datetime.now(timezone.utc):%Y-%m-%d %H:%M:%S UTC}",
                f"Duration:  {time.time() - started:.1f}s",
                f"Load ID:   {load_info.loads_ids[-1]}",
            ]),
        )
    except Exception as e:
        try:
            send_email(
                f"[dltHub] {pipeline.pipeline_name} FAILED",
                f"Pipeline:  {pipeline.pipeline_name}\nError:     {type(e).__name__}: {e}",
            )
        except Exception as mail_err:
            print(f"Failed to send failure email: {mail_err}")
        raise

Wrap the failure-path send_email in its own try/except: a broken alerting channel shouldn't mask the underlying pipeline error.

Deploy and trigger

uv run dlthub deploy                            # syncs code + SMTP credentials
uv run dlthub run my_job                        # triggers the job, email lands on completion