README.md

Installation

pip install pipen-email

Quickstart

from pipen import Pipen, Proc

class SayHello(Proc):
    input = "name:var"
    output = "outfile:file:greeting.txt"
    script = 'echo "Hello, {{in.name}}!" > {{out.outfile}}'

pipeline = Pipen(
    name="my-pipeline",
    loglevel="debug",
    plugin_opts={
        # SMTP — email_host is required for any notifications to fire
        "email_host": "smtp.example.com",
        "email_port": 587,
        "email_use_tls": True,
        "email_username": "user",
        "email_password": "pass",
        # Envelope
        "email_to": "you@example.com",
        # Job digest (opt-in)
        "email_on_job": True,
        "email_batch_interval": 60,
    },
)
pipeline.set_start(SayHello).set_data(["Alice", "Bob"]).run()

Note: email_host must be set to a truthy value. Without it, no emails will be sent — all notification keys are silently ignored. This acts as a master switch so you can disable the plugin entirely by omitting email_host.

SMTP Examples

Local debug server

# Terminal 1 — start a debug SMTP server
python -m aiosmtpd -n -l localhost:1025

# Terminal 2 — run the pipeline
python my_pipeline.py

plugin_opts = {
    "email_host": "localhost",
    "email_port": 1025,
    "email_to": "dev@localhost",
}

Gmail (with App Password)

Generate an app password at myaccount.google.com/apppasswords.

plugin_opts = {
    "email_host": "smtp.gmail.com",
    "email_port": 587,
    "email_use_tls": True,
    "email_username": "you@gmail.com",
    "email_password": "your-16-char-app-password",
    "email_to": "you@gmail.com",
}

No-auth relay

plugin_opts = {
    "email_host": "smtp-relay.internal",
    "email_port": 25,
    "email_to": "team@example.com",
}

AWS SES

plugin_opts = {
    "email_host": "email-smtp.us-east-1.amazonaws.com",
    "email_port": 587,
    "email_use_tls": True,
    "email_username": "AKIA...",
    "email_password": "your-ses-smtp-password",
    "email_to": "team@example.com",
}

Configuration Reference

All options live in pipen.config.plugin_opts. Set them via Pipen(plugin_opts={...}), .pipen.toml, or per-process Proc.plugin_opts.

SMTP connection

Key	Default	Description
email_host	None	SMTP server hostname. Must be set for any emails to send.
email_port	25	SMTP server port
email_use_tls	False	Use STARTTLS on connect
email_use_ssl	False	Use SMTPS (SSL on port 465)
email_username	None	SMTP auth username
email_password	None	SMTP auth password
email_timeout	30	SMTP connection timeout in seconds

Email envelope

Key	Default	Description
email_from	"pipen@localhost"	From address
email_to	None	To address(es) — comma-separated string or list
email_cc	None	CC address(es) — comma-separated string or list
email_bcc	None	BCC address(es) — comma-separated string or list
email_subject_prefix	"[pipen]"	Prefix prepended to every subject line

Pipeline lifecycle

Key	Default	Description
email_on_start	True	Send email when pipeline starts
email_on_complete	True	Send email when pipeline finishes (success or failure)

Process lifecycle

Key	Default	Description
email_on_proc_start	True	Send email when a process starts
email_on_proc_done	True	Send email when a process completes, fails, or returns cached
email_on_proc_shutdown	True	Send email when a process receives a signal (SIGTERM, SIGKILL, etc.)

Note: email_on_proc_done covers all terminal statuses — completed, failed, and cached. There is no separate email_on_proc_failed toggle. When a process fails, the email body includes the failed job's script path, stdout/stderr file paths, and full stderr content.

Job lifecycle

Key	Default	Description
email_on_job	False	Enable batched job status digest emails
email_batch_interval	60	Minimum seconds between job digest emails

Logging

Key	Default	Description
email_loglevel	"info"	Log level for email send events. One of "debug", "info", "warning", "error", or "critical". Case-insensitive.

How It Works

Email guard

All hooks check email_host first via the internal _should_send method. If email_host is not set (or falsy), every notification silently skips. This means you can conditionally enable the plugin:

# Disable in dev, enable in CI
plugin_opts = {
    "email_host": "smtp.example.com" if os.environ.get("CI") else None,
    "email_to": "team@example.com",
}

Pipeline emails

Sent on on_start and on_complete. Body format:

Pipeline: my-pipeline
Status: COMPLETED
Workdir: .pipen/my-pipeline
Outdir: /output/path
Processes: 3
Process names: ProcA, ProcB, ProcC

Process emails

Sent on on_proc_start, on_proc_done, and on_proc_shutdown. Body format:

Pipeline: my-pipeline
Process: ProcA
Status: COMPLETED
Jobs: 10
Workdir: .pipen/my-pipeline/ProcA

Job status summary:
INIT: 0-9
SUCCEEDED: 0-9

On process failure, the body also includes details about the first failed job:

Information for job #3:
- script: .pipen/my-pipeline/ProcA/3/job.script
- stdout: .pipen/my-pipeline/ProcA/3/job.stdout
- stderr: .pipen/my-pipeline/ProcA/3/job.stderr

Full STDERR:
----------------
Traceback (most recent call last):
...

Job digest emails

When email_on_job is True, each job lifecycle event records its index. A digest email is sent when email_batch_interval seconds have elapsed since the last digest. Digest body format:

Job status digest:
---------------------
INIT: 0-9
RUNNING: 0-9
SUCCEEDED: 0-2
FAILED: 3-5

Job digest emails are also flushed automatically in on_proc_done (via the interval gate).

Shutdown email

Only sent when the process receives an actual signal (sig is truthy). For normal process completion, on_proc_shutdown is called with sig=None and no email is sent — the on_proc_done email covers that case.

Error Handling

SMTP errors are caught and logged at ERROR level. The pipeline never fails due to an email error — _send_email returns False on failure, and the caller continues.

See Also

• pipen — pipeline framework
• example.py — runnable smoke test