Zum Hauptinhalt springen

SES

Transactional Email

Amazon Simple Email Service (SES) sends transactional emails.

Email Types

TypeExamplePriority
ReceiptsDonation confirmationHigh
NotificationsCampaign updatesMedium
AccountPassword reset, verificationHigh
ReportsWeekly transparency reportsLow

Configuration

Domain Verification

Domain: amply-impact.org
DKIM: Enabled
SPF: Configured
DMARC: Configured

DNS records:

# DKIM
selector1._domainkey.amply-impact.org CNAME selector1.dkim.amazonses.com
selector2._domainkey.amply-impact.org CNAME selector2.dkim.amazonses.com
selector3._domainkey.amply-impact.org CNAME selector3.dkim.amazonses.com

# SPF (add to existing)
amply-impact.org TXT "v=spf1 include:amazonses.com ~all"

# DMARC
_dmarc.amply-impact.org TXT "v=DMARC1; p=quarantine; rua=mailto:dmarc@amply-impact.org"

Sending Identity

IdentityType: DOMAIN
Identity: amply-impact.org
ConfigurationSetName: amply-emails

Configuration Set

ConfigurationSetName: amply-emails

EventDestinations:
# Track bounces and complaints
- EventDestinationName: feedback
Enabled: true
MatchingEventTypes:
- BOUNCE
- COMPLAINT
CloudWatchDestination:
DimensionConfigurations:
- DimensionName: EmailType
DimensionValueSource: MESSAGE_TAG

Email Templates

Donation Receipt

{
"TemplateName": "donation-receipt",
"SubjectPart": "Thank you for your donation to {{organisation_name}}",
"HtmlPart": "<!DOCTYPE html>...",
"TextPart": "Thank you for your donation..."
}

Password Reset

{
"TemplateName": "password-reset",
"SubjectPart": "Reset your Amply password",
"HtmlPart": "...",
"TextPart": "..."
}

Implementation

Email Service

# lib/email.py
import boto3
from jinja2 import Environment, PackageLoader

ses = boto3.client('ses', region_name='eu-central-1')
jinja = Environment(loader=PackageLoader('amply', 'templates/email'))

class EmailService:
FROM_ADDRESS = "Amply <noreply@amply-impact.org>"

async def send_email(
self,
to: str,
template: str,
context: dict,
tags: dict | None = None
) -> str:
"""Send templated email."""
# Render templates
html_template = jinja.get_template(f"{template}.html")
text_template = jinja.get_template(f"{template}.txt")

html_body = html_template.render(**context)
text_body = text_template.render(**context)

# Get subject from template or context
subject = context.get('subject', self._get_subject(template, context))

response = ses.send_email(
Source=self.FROM_ADDRESS,
Destination={'ToAddresses': [to]},
Message={
'Subject': {'Data': subject},
'Body': {
'Html': {'Data': html_body},
'Text': {'Data': text_body}
}
},
ConfigurationSetName='amply-emails',
Tags=[
{'Name': 'EmailType', 'Value': template},
*[{'Name': k, 'Value': v} for k, v in (tags or {}).items()]
]
)

return response['MessageId']

async def send_donation_receipt(
self,
to: str,
donation: Donation,
organisation: Organisation
) -> str:
"""Send donation receipt."""
return await self.send_email(
to=to,
template='donation-receipt',
context={
'donor_name': donation.donor_name,
'amount': format_currency(donation.amount, donation.currency),
'organisation_name': organisation.name,
'donation_id': donation.id,
'date': donation.created_at.strftime('%Y-%m-%d'),
'verify_url': f"https://amply-impact.org/public/verify/{donation.id}"
},
tags={'donation_id': donation.id}
)

Celery Tasks

# jobs/email.py
from amply.lib.email import EmailService

email_service = EmailService()

@app.task(bind=True, max_retries=3)
def send_email(
self,
to: str,
template: str,
context: dict
) -> str:
"""Send email via SES."""
try:
return email_service.send_email(to, template, context)
except Exception as e:
raise self.retry(exc=e, countdown=60 * (self.request.retries + 1))

@app.task
def send_donation_receipt(donation_id: str):
"""Send donation receipt email."""
donation = get_donation(donation_id)
organisation = get_organisation(donation.organisation_id)
donor = get_donor(donation.donor_id)

if donor and donor.email:
email_service.send_donation_receipt(
to=donor.email,
donation=donation,
organisation=organisation
)

Templates

Template Structure

src/amply/templates/email/
├── base.html # Base HTML template
├── base.txt # Base text template
├── donation-receipt.html
├── donation-receipt.txt
├── password-reset.html
├── password-reset.txt
├── campaign-update.html
├── campaign-update.txt
├── recurring-reminder.html
└── recurring-reminder.txt

Base Template

<!-- base.html -->
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<style>
body { font-family: -apple-system, BlinkMacSystemFont, sans-serif; }
.container { max-width: 600px; margin: 0 auto; }
.header { background: #000; color: #fff; padding: 20px; }
.content { padding: 20px; }
.footer { padding: 20px; color: #666; font-size: 12px; }
</style>
</head>
<body>
<div class="container">
<div class="header">
<img src="https://amply-impact.org/logo.png" alt="Amply" height="30">
</div>
<div class="content">
{% block content %}{% endblock %}
</div>
<div class="footer">
{% block footer %}
<p>Amply - Ultra-transparent charitable giving</p>
<p><a href="https://amply-impact.org">amply-impact.org</a></p>
{% endblock %}
</div>
</div>
</body>
</html>

Bounce/Complaint Handling

SNS Notification

# SNS topic for SES feedback
TopicName: amply-ses-feedback

Subscription:
Protocol: https
Endpoint: https://api.amply-impact.org/v1/webhooks/ses-feedback

Webhook Handler

@router.post("/webhooks/ses-feedback")
async def ses_feedback_webhook(request: Request):
"""Handle SES bounce/complaint notifications."""
body = await request.json()

message_type = body.get('Type')

if message_type == 'SubscriptionConfirmation':
# Confirm SNS subscription
requests.get(body['SubscribeURL'])
return {"status": "confirmed"}

if message_type == 'Notification':
message = json.loads(body['Message'])
notification_type = message.get('notificationType')

if notification_type == 'Bounce':
await handle_bounce(message['bounce'])
elif notification_type == 'Complaint':
await handle_complaint(message['complaint'])

return {"status": "ok"}

async def handle_bounce(bounce: dict):
"""Handle bounced email."""
for recipient in bounce.get('bouncedRecipients', []):
email = recipient['emailAddress']
# Mark email as invalid
await mark_email_bounced(email)

async def handle_complaint(complaint: dict):
"""Handle spam complaint."""
for recipient in complaint.get('complainedRecipients', []):
email = recipient['emailAddress']
# Unsubscribe user
await unsubscribe_email(email)

Monitoring

Metrics

  • Send rate
  • Bounce rate
  • Complaint rate
  • Delivery rate

Alarms

- AlarmName: ses-high-bounce-rate
MetricName: Reputation.BounceRate
Threshold: 0.05 # 5%

- AlarmName: ses-high-complaint-rate
MetricName: Reputation.ComplaintRate
Threshold: 0.001 # 0.1%

Cost

  • First 62,000 emails/month: Free (from EC2/Lambda)
  • After: $0.10 per 1,000 emails

Estimated: ~$5/month at typical volume


Related: