AMSDAL Mail

Universal email integration plugin for AMSDAL Framework. Provides a unified interface for sending emails through multiple backends (SMTP, AWS SES, etc.).

Features

• Multiple Backends: SMTP, AWS SES, Console, Dummy
• Unified API: Single interface for all email services
• Async Support: Native async/await support for all backends
• Template Support: Send templated emails with variable substitution (SES)
• Tags & Metadata: Tag and track emails for analytics and filtering
• Click/Open Tracking: Monitor email opens and link clicks
• Inline Images: Embed images in HTML via CID references
• Type Safe: Full Pydantic validation with type hints
• Django-like API: Familiar interface for Django developers
• AMSDAL Integration: Seamless integration with AMSDAL Framework lifecycle
• Extensible: Easy to add custom backends

Documentation

• Quick Start - Get started quickly
• Configuration - Backend and environment setup
• SMTP Usage Guide - Detailed SMTP examples (Gmail, etc.)
• Architecture - System design and patterns
• Tracking Guide - Email click/open tracking setup
• Webhooks - Receiving tracking events from ESPs

Installation

# Basic installation
pip install amsdal-mail

# With specific backend support
pip install amsdal-mail[smtp]
pip install amsdal-mail[ses]

# All backends
pip install amsdal-mail[all]

Quick Start

Basic Usage

from amsdal_mail import send_mail

# Send a simple email
status = send_mail(
    subject='Hello from AMSDAL Mail',
    message='This is a test email',
    from_email='sender@example.com',
    recipient_list=['recipient@example.com'],
)

print(status.is_success)  # True

Async Usage

import asyncio
from amsdal_mail import asend_mail

async def send_async_email():
    status = await asend_mail(
        subject='Async Email',
        message='Sent asynchronously',
        from_email='sender@example.com',
        recipient_list=['recipient@example.com'],
    )
    print(status.is_success)

asyncio.run(send_async_email())

HTML Emails

from amsdal_mail import send_mail

send_mail(
    subject='HTML Email',
    message='Plain text version',
    html_message='<h1>HTML version</h1>',
    from_email='sender@example.com',
    recipient_list=['recipient@example.com'],
)

Advanced Usage

from amsdal_mail import get_connection, EmailMessage, Attachment

# Get connection to specific backend
with get_connection('smtp') as conn:
    # Create message with attachments
    message = EmailMessage(
        subject='Email with Attachment',
        body='See attached document',
        from_email='sender@example.com',
        to=['recipient@example.com'],
        cc=['cc@example.com'],
        bcc=['bcc@example.com'],
        reply_to=['reply@example.com'],
        attachments=[
            Attachment(
                filename='document.pdf',
                content=b'PDF content...',
                mimetype='application/pdf',
            )
        ],
        headers={'X-Custom-Header': 'value'},
    )

    # Send multiple messages in one connection
    status = conn.send_messages([message])

Template Support

Send emails using ESP templates with variable substitution:

from amsdal_mail import EmailMessage, get_connection

# Basic template with global variables
message = EmailMessage(
    subject='Welcome!',
    from_email='noreply@example.com',
    to=['user@example.com'],
    template_id='welcome-template',
    merge_global_data={
        'company': 'Acme Inc',
        'year': '2024',
    },
)

# Template with per-recipient variables
message = EmailMessage(
    subject='Order Confirmation',
    from_email='orders@example.com',
    to=['customer@example.com'],
    template_id='order-confirmation',
    merge_data={
        'customer@example.com': {
            'name': 'John Doe',
            'order_id': '12345',
            'total': '$99.99',
        },
    },
    merge_global_data={
        'company': 'Acme Inc',
        'support_email': 'support@example.com',
    },
)

backend = get_connection('ses')
backend.send_messages([message])

Note: Template support is currently available for AWS SES backend. Templates must be created in your ESP dashboard first.

Tags and Metadata

Add tags and metadata for tracking and analytics:

from amsdal_mail import EmailMessage

message = EmailMessage(
    subject='Marketing Newsletter',
    body='Newsletter content...',
    from_email='marketing@example.com',
    to=['subscriber@example.com'],
    tags=['newsletter', 'marketing', 'q4-2024'],
    metadata={
        'campaign_id': 'fall-sale-2024',
        'user_id': '12345',
        'ab_test_variant': 'A',
    },
)

Tags and metadata are passed to the ESP and can be used for:

• Filtering and categorizing emails
• Analytics and reporting
• A/B testing tracking
• Custom event data

Click and Open Tracking

Enable tracking to monitor email opens and link clicks:

from amsdal_mail import EmailMessage, get_connection

message = EmailMessage(
    subject='Product Launch',
    body='Check out our new product!',
    html_body='<h1>New Product</h1><a href="https://example.com/product">Learn More</a>',
    from_email='marketing@example.com',
    to=['customer@example.com'],
    track_opens=True,   # Track when email is opened
    track_clicks=True,  # Track when links are clicked
)

backend = get_connection('ses')
backend.send_messages([message])

Important Notes:

• Open tracking only works for HTML emails (inserts invisible tracking pixel)
• Click tracking rewrites URLs to track clicks before redirecting
• AWS SES: Requires Configuration Set setup (see docs/TRACKING.md)

For detailed tracking setup instructions, see docs/TRACKING.md.

Inline Images

Embed images directly in HTML emails using CID references:

from amsdal_mail import EmailMessage, Attachment, get_connection

message = EmailMessage(
    subject='Email with Embedded Image',
    body='Please enable HTML to view this email.',
    html_body='<img src="cid:logo"><h1>Welcome!</h1>',
    from_email='sender@example.com',
    to=['recipient@example.com'],
    attachments=[
        Attachment(
            filename='logo.png',
            content=open('logo.png', 'rb').read(),
            mimetype='image/png',
            content_id='logo',  # Referenced as cid:logo in HTML
        ),
    ],
)

connection = get_connection('smtp')
status = connection.send_messages([message])

For more examples see docs/SMTP_USAGE.md.

Configuration

Environment Variables

Backend Selection

AMSDAL_EMAIL_BACKEND=smtp  # console, smtp, dummy, ses

SMTP Configuration

AMSDAL_EMAIL_HOST=smtp.gmail.com
AMSDAL_EMAIL_PORT=587
AMSDAL_EMAIL_USER=your-email@gmail.com
AMSDAL_EMAIL_PASSWORD=your-password
AMSDAL_EMAIL_USE_TLS=true
AMSDAL_EMAIL_USE_SSL=false
AMSDAL_EMAIL_TIMEOUT=30

AWS SES Configuration

AMSDAL_EMAIL_BACKEND=ses
AWS_ACCESS_KEY_ID=your-access-key
AWS_SECRET_ACCESS_KEY=your-secret-key
AWS_REGION=us-east-1

Programmatic Configuration

from amsdal_mail import get_connection

# Override configuration
connection = get_connection(
    backend='smtp',
    host='smtp.example.com',
    port=587,
    username='user@example.com',
    password='secret',
    use_tls=True,
)

connection.send_messages([message])

Backends

Console Backend

Purpose: Development and debugging

from amsdal_mail import send_mail

# Outputs to stdout
send_mail(
    subject='Test',
    message='This will print to console',
    from_email='sender@example.com',
    recipient_list=['recipient@example.com'],
)

Output:

Subject: Test
From: sender@example.com
To: recipient@example.com
Body:
This will print to console
----------------------------------------

SMTP Backend

Purpose: Generic SMTP servers (Gmail, Outlook, etc.)

# Configuration
export AMSDAL_EMAIL_BACKEND=smtp
export AMSDAL_EMAIL_HOST=smtp.gmail.com
export AMSDAL_EMAIL_PORT=587
export AMSDAL_EMAIL_USER=your-email@gmail.com
export AMSDAL_EMAIL_PASSWORD=your-app-password
export AMSDAL_EMAIL_USE_TLS=true

Gmail Example:

1. Enable 2FA in Google Account
2. Generate App Password
3. Use App Password in AMSDAL_EMAIL_PASSWORD

For detailed SMTP examples, see docs/SMTP_USAGE.md.

AWS SES Backend

Purpose: Amazon Simple Email Service

export AMSDAL_EMAIL_BACKEND=ses
export AWS_ACCESS_KEY_ID=your-key
export AWS_SECRET_ACCESS_KEY=your-secret
export AWS_REGION=us-east-1

Dummy Backend

Purpose: Testing (no actual sending)

from amsdal_mail import get_connection

# Nothing is sent, but returns success status
connection = get_connection('dummy')
status = connection.send_messages([message])
print(status.is_success)  # True

AMSDAL Framework Integration

Register Plugin

Add to your AMSDAL application configuration:

# settings.py or .env
AMSDAL_CONTRIBS = 'amsdal_mail.app.MailAppConfig'

Or multiple plugins:

# .env
AMSDAL_CONTRIBS=amsdal.contrib.auth.app.AuthAppConfig,amsdal_mail.app.MailAppConfig

Use in AMSDAL App

from amsdal_mail import send_mail

# Now available throughout your AMSDAL application
send_mail(
    subject='Welcome',
    message='Thanks for signing up!',
    from_email='noreply@example.com',
    recipient_list=[user.email],
)

API Reference

send_mail()

def send_mail(
    subject: str,
    message: str,
    from_email: str,
    recipient_list: list[str] | str,
    fail_silently: bool = False,
    html_message: str | None = None,
    connection = None,
    **kwargs,
) -> SendStatus

Send a single email message.

Parameters:

• subject: Email subject line
• message: Plain text body
• from_email: Sender email address
• recipient_list: List of recipient email addresses (or single string)
• fail_silently: If True, suppress exceptions and return empty SendStatus
• html_message: HTML version of body (optional)
• connection: Reuse existing connection (optional)
• **kwargs: Additional arguments passed to EmailMessage

Returns: SendStatus object with message IDs, statuses, and ESP response details

Example:

status = send_mail(
    'Welcome',
    'Hello!',
    'noreply@example.com',
    ['user@example.com'],
)

# Access send details
print(status.message_id)          # ESP message ID (e.g., 'msg-123')
print(status.is_success)          # True if all sent successfully
print(status.status)              # {'sent'}
print(status.recipients)          # Per-recipient details

asend_mail()

async def asend_mail(...) -> SendStatus

Async version of send_mail(). Returns SendStatus object with send details.

get_connection()

def get_connection(
    backend: str | None = None,
    fail_silently: bool = False,
    **kwargs,
)

Get email backend connection.

Parameters:

• backend: Backend name ('smtp', 'ses', 'console', 'dummy') or full class path
• fail_silently: Suppress errors if True
• **kwargs: Backend-specific configuration

Returns: Backend instance

EmailMessage

from amsdal_mail import EmailMessage, Attachment

message = EmailMessage(
    subject: str,
    body: str,
    from_email: EmailStr | str,
    to: list[EmailStr | str],
    cc: list[EmailStr | str] = [],
    bcc: list[EmailStr | str] = [],
    reply_to: list[EmailStr | str] = [],
    attachments: list[Attachment] = [],
    headers: dict[str, str] = {},
    html_body: str | None = None,
)

Pydantic model representing an email message.

Attachment

from amsdal_mail import Attachment

attachment = Attachment(
    filename: str,              # File name
    content: bytes,             # File content
    mimetype: str,              # MIME type (e.g., 'application/pdf')
    content_id: str | None,     # Content-ID for inline images (e.g., 'logo' for cid:logo)
)

SendStatus

from amsdal_mail import SendStatus, RecipientStatus

# Returned by send_mail() and asend_mail()
status = send_mail(...)

# SendStatus attributes:
status.message_id          # str | set[str] | None - ESP message ID(s)
status.status              # set[SendStatusType] | None - Set of statuses
status.recipients          # dict[str, RecipientStatus] - Per-recipient details
status.esp_response        # Any - Raw ESP API response

# Helper methods:
status.is_success                    # True if all sent/queued
status.has_failures                  # True if any failed/rejected/invalid
status.get_successful_recipients()   # List of successful emails
status.get_failed_recipients()       # List of failed emails

# RecipientStatus attributes:
recipient = status.recipients['user@example.com']
recipient.message_id       # str | None - ESP message ID for this recipient
recipient.status          # 'sent' | 'queued' | 'failed' | 'rejected' | 'invalid' | 'unknown'

Status Types:

• sent - ESP has sent the message (queued for delivery)
• queued - ESP will try to send later
• invalid - Recipient email address is not valid
• rejected - Recipient is blacklisted or rejected by ESP
• failed - Send attempt failed for some other reason
• unknown - Status could not be determined

Testing

Using Console Backend

# In your test configuration
import os
os.environ['AMSDAL_EMAIL_BACKEND'] = 'console'

# Emails will print to stdout
from amsdal_mail import send_mail
send_mail(...)  # Prints to console

Using Dummy Backend

import os
os.environ['AMSDAL_EMAIL_BACKEND'] = 'dummy'

# Emails are "sent" but do nothing
from amsdal_mail import send_mail
status = send_mail(...)  # Returns SendStatus, does nothing
print(status.is_success)  # True

Mocking in Tests

import pytest
from amsdal_mail import send_mail, SendStatus

def test_email_sending(mocker):
    # Mock the backend
    mock_backend = mocker.patch('amsdal_mail.backends.get_connection')
    mock_backend.return_value.send_messages.return_value = SendStatus()

    # Test your code
    result = send_mail(
        subject='Test',
        message='Test message',
        from_email='sender@example.com',
        recipient_list=['recipient@example.com'],
    )

    mock_backend.return_value.send_messages.assert_called_once()

Creating Custom Backends

Define Backend Class

from amsdal_mail.backends.base import BaseEmailBackend
from amsdal_mail import EmailMessage, SendStatus

class MyCustomBackend(BaseEmailBackend):
    def __init__(self, api_key: str, **kwargs):
        super().__init__(**kwargs)
        self.api_key = api_key

    def send_messages(self, email_messages: list[EmailMessage]) -> SendStatus:
        status = SendStatus()
        # Your custom sending logic
        return status

    async def asend_messages(self, email_messages: list[EmailMessage]) -> SendStatus:
        status = SendStatus()
        # Async implementation
        return status

Register Backend

from amsdal_mail.backends import BACKENDS
BACKENDS['mycustom'] = 'myapp.backends.MyCustomBackend'

# Now you can use it
from amsdal_mail import get_connection
connection = get_connection('mycustom', api_key='secret')

Or use full import path:

connection = get_connection('myapp.backends.MyCustomBackend', api_key='secret')

Examples

Bulk Email Sending

from amsdal_mail import get_connection, EmailMessage

def send_bulk_emails(users: list):
    # Reuse connection for efficiency
    with get_connection('smtp') as conn:
        messages = [
            EmailMessage(
                subject=f'Hello {user.name}',
                body=f'Welcome {user.name}!',
                from_email='noreply@example.com',
                to=[user.email],
            )
            for user in users
        ]

        # Send all at once
        status = conn.send_messages(messages)
        print(f'Success: {status.is_success}')

Transactional Email

from amsdal_mail import send_mail

def send_password_reset(user, reset_link: str):
    send_mail(
        subject='Password Reset Request',
        message=f'''
        Hello {user.name},

        You requested a password reset. Click the link below:
        {reset_link}

        If you didn't request this, ignore this email.
        ''',
        html_message=f'''
        <h2>Password Reset Request</h2>
        <p>Hello {user.name},</p>
        <p>You requested a password reset.</p>
        <a href="{reset_link}">Reset Password</a>
        <p>If you didn't request this, ignore this email.</p>
        ''',
        from_email='noreply@example.com',
        recipient_list=[user.email],
    )

Async Bulk Sending

import asyncio
from amsdal_mail import asend_mail

async def send_notifications(users: list):
    tasks = [
        asend_mail(
            subject='New Update',
            message=f'Hi {user.name}, check out our new features!',
            from_email='notifications@example.com',
            recipient_list=[user.email],
        )
        for user in users
    ]

    # Send concurrently
    results = await asyncio.gather(*tasks)
    print(f'All succeeded: {all(s.is_success for s in results)}')

# Run
asyncio.run(send_notifications(users))

Troubleshooting

Gmail SMTP Issues

Error: "Username and Password not accepted"

Solution:

1. Enable 2-Factor Authentication
2. Generate App Password (Settings > Security > App Passwords)
3. Use App Password instead of regular password

TLS/SSL Issues

Error: "STARTTLS extension not supported"

Solution:

# Try different port and TLS settings
AMSDAL_EMAIL_PORT=465
AMSDAL_EMAIL_USE_SSL=true
AMSDAL_EMAIL_USE_TLS=false

Connection Timeout

Error: "Connection timed out"

Solution:

# Increase timeout
AMSDAL_EMAIL_TIMEOUT=60

Development

Setup

git clone <repo-url>
cd amsdal_mail

# Install dependencies
pip install uv hatch
hatch env create
hatch run sync

# Run checks
hatch run all      # style + typing
hatch run test     # tests
hatch run cov      # tests with coverage

Release Workflow

1. Develop on a feature branch, create PR to main — CI runs lint + tests
2. When ready to release, create a release/X.Y.Z branch, bump version in amsdal_mail/__about__.py, update CHANGELOG.md
3. Merge release/* to main — CD workflow automatically creates tag, builds, publishes to PyPI, and creates GitHub Release with changelog