Mailtrap Airflow Provider

An Apache Airflow provider for Mailtrap.io - send transactional emails from your Airflow DAGs.

Installation

pip install airflow-provider-mailtrap

Requirements

• Python >= 3.9
• Apache Airflow >= 2.4
• Mailtrap API token (get one here)

Setup

1. Create a Mailtrap Connection

In the Airflow UI, go to Admin → Connections → Add:

2. Use the Operator in Your DAG

from mailtrap_provider.operators.send_email import MailtrapSendEmailOperator

send_email = MailtrapSendEmailOperator(
    task_id="send_welcome_email",
    to="user@example.com",
    subject="Welcome!",
    html="<h1>Welcome to our service!</h1>",
    sender="noreply@yourdomain.com",
    sender_name="My App",
)

Usage Examples

Send a Plain Text Email

from mailtrap_provider.operators.send_email import MailtrapSendEmailOperator

send_notification = MailtrapSendEmailOperator(
    task_id="send_notification",
    to="user@example.com",
    subject="Task Completed",
    text="Your workflow has completed successfully.",
    sender="notifications@yourdomain.com",
)

Send an HTML Email

send_report = MailtrapSendEmailOperator(
    task_id="send_report",
    to="team@example.com",
    subject="Daily Report",
    html="""
    <html>
        <body>
            <h1>Daily Report</h1>
            <p>Here is your daily summary...</p>
        </body>
    </html>
    """,
    sender="reports@yourdomain.com",
    sender_name="Report Bot",
    category="daily-reports",  # For Mailtrap analytics
)

Send to Multiple Recipients

send_to_team = MailtrapSendEmailOperator(
    task_id="notify_team",
    to=["alice@example.com", "bob@example.com"],
    subject="Team Update",
    text="Important update for the team.",
    sender="noreply@yourdomain.com",
)

Use Templated Fields

All parameters are templatable, so you can use Jinja templates:

send_dynamic_email = MailtrapSendEmailOperator(
    task_id="send_dynamic_email",
    to="{{ var.value.recipient_email }}",
    subject="Report for {{ ds }}",
    text="Data processed: {{ ti.xcom_pull(task_ids='process_data') }}",
    sender="{{ var.value.sender_email }}",
)

Operator Parameters

Using the Hook Directly

For more advanced use cases, you can use the hook directly:

from mailtrap_provider.hooks.mailtrap import MailtrapHook

# Simple usage with the helper method
hook = MailtrapHook()
response = hook.send_email(
    sender="sender@example.com",
    to="recipient@example.com",
    subject="Hello",
    text="Hello from Airflow!",
)

# Advanced usage with raw client
client = hook.get_conn()  # Returns mailtrap.MailtrapClient
# Use client directly for advanced features

Connection Extras

You can store default sender information in the connection extras:

{
  "sender": "noreply@yourdomain.com",
  "sender_name": "Your Application"
}

When sender is not provided to the operator, it will use the value from connection extras.

Development

Setup Development Environment

# Install Astro CLI (macOS)
brew install astro

# Setup dev environment
./scripts/setup-dev.sh

# Start local Airflow
cd dev && astro dev start

Access Airflow UI at http://localhost:8080 (user: admin, pass: admin)

Note: The setup script enables AIRFLOW__CORE__TEST_CONNECTION=Enabled so the "Test" button works in the connection form. This is disabled by default in Airflow for security.

After Code Changes

./scripts/sync-provider.sh

Run Tests

pip install pytest requests-mock
pytest tests/ -v

License

Apache License 2.0 - see LICENSE for details.

Links

• Mailtrap Documentation
• Mailtrap Python SDK
• Apache Airflow