SendGrid: Transactional Email API Guide

# setup.py
# Initialize SendGrid client

import os
from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Mail, Email, To, Content

SENDGRID_API_KEY = os.environ.get('SENDGRID_API_KEY')
FROM_EMAIL = 'noreply@dodatech.com'
FROM_NAME = 'DodaTech Security'

sg = SendGridAPIClient(SENDGRID_API_KEY)

# send_email.py
# Send a basic transactional email via SendGrid REST API

from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Mail
import os

def send_basic_email(to_email, subject, body):
    """Send a plain text email via SendGrid REST API."""
    message = Mail(
        from_email='noreply@dodatech.com',
        to_emails=to_email,
        subject=subject,
        plain_text_content=body,
    )

    try:
        response = sg.send(message)
        print(f"Email sent to {to_email}")
        print(f"Status: {response.status_code}")
        print(f"Headers: {response.headers}")

        # SendGrid returns 202 Accepted on success
        if response.status_code == 202:
            print("✅ Email queued for delivery")
        return response

    except Exception as e:
        print(f"❌ Failed to send: {e}")
        return None

send_basic_email(
    to_email='alice@example.com',
    subject='Your Doda Browser Security Report',
    body='Hi Alice,\n\nYour weekly security report is ready. No threats detected in the past 7 days.\n\nBest,\nThe Doda Team'
)

# template_email.py
# Send email using SendGrid dynamic templates with Handlebars

from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Mail, TemplateId
import os

def send_template_email(to_email, template_id, template_data):
    """Send an email using a dynamic template."""
    message = Mail(
        from_email='noreply@dodatech.com',
        to_emails=to_email,
    )
    message.template_id = TemplateId(template_id)

    # Pass dynamic data to the template
    # Template uses {{ name }}, {{ report_url }}, etc.
    message.dynamic_template_data = template_data

    try:
        response = sg.send(message)
        if response.status_code == 202:
            print(f"Template email sent to {to_email}")
            print(f"Template: {template_id}")
        return response

    except Exception as e:
        print(f"Template email failed: {e}")
        body = e.body if hasattr(e, 'body') else str(e)
        print(f"Error body: {body}")
        return None

# Design this template in SendGrid's Template Editor with:
# Subject: Your {{ report_type }} is ready
# Body: Hi {{ name }}, your {{ report_type }} is ready.
#       Threats found: {{ threat_count }}. View report: {{ report_url }}

send_template_email(
    to_email='bob@example.com',
    template_id='d-abc123def456',  # Create in SendGrid Dashboard
    template_data={
        'name': 'Bob',
        'report_type': 'Weekly Security Report',
        'threat_count': 0,
        'report_url': 'https://dodatech.com/reports/weekly-2026-06-20',
    }
)

# attachment_email.py
# Send email with file attachments

from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Mail, Attachment, FileContent, FileName, FileType, Disposition
import os
import base64

def send_email_with_attachment(to_email, subject, body_html, file_path):
    """Send an email with a file attachment."""
    message = Mail(
        from_email='reports@dodatech.com',
        to_emails=to_email,
        subject=subject,
        html_content=body_html,
    )

    # Read and encode the file
    with open(file_path, 'rb') as f:
        data = f.read()
        encoded = base64.b64encode(data).decode()

    attachment = Attachment(
        FileContent(encoded),
        FileName(os.path.basename(file_path)),
        FileType('application/pdf'),
        Disposition('attachment'),
    )
    message.attachment = attachment

    # Send
    response = sg.send(message)
    print(f"Email with attachment sent to {to_email}")
    print(f"File: {os.path.basename(file_path)}")
    print(f"Status: {response.status_code}")

    return response

# send_email_with_attachment(
#     to_email='alice@example.com',
#     subject='Your Security Report PDF',
#     body_html='<h1>Your Report</h1><p>See attached PDF.</p>',
#     file_path='/tmp/report_2026-06-20.pdf'
# )

# batch_email.py
# Send personalized emails to multiple recipients

from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Mail, Personalization, Email, Substitution
import os

def send_batch_emails(recipients, template_id):
    """Send personalized emails to multiple recipients.
    
    Args:
        recipients: List of dicts with email, name, and custom fields
        template_id: SendGrid dynamic template ID
    """
    message = Mail()
    message.from_email = Email('noreply@dodatech.com', 'DodaTech')
    message.template_id = TemplateId(template_id)

    # Add each recipient with personalized data
    for recipient in recipients:
        personalization = Personalization()
        personalization.add_to(Email(recipient['email']))
        personalization.dynamic_template_data = recipient['data']
        message.add_personalization(personalization)

    try:
        response = sg.send(message)
        print(f"Batch sent to {len(recipients)} recipients")
        print(f"Status: {response.status_code}")

        # Check for errors in individual deliveries
        if response.status_code == 202:
            print("✅ All emails accepted for delivery")
        return response

    except Exception as e:
        print(f"Batch send failed: {e}")
        return None

# Example: send weekly report to all users
users = [
    {'email': 'alice@example.com', 'data': {'name': 'Alice', 'threats': 0}},
    {'email': 'bob@example.com', 'data': {'name': 'Bob', 'threats': 2}},
    {'email': 'carol@example.com', 'data': {'name': 'Carol', 'threats': 0}},
]

send_batch_emails(users, template_id='d-abc123def456')

# webhooks.py
# Handle SendGrid event webhooks (opens, clicks, bounces)

from flask import Flask, request, jsonify
import json

app = Flask(__name__)

@app.route('/sendgrid/events', methods=['POST'])
def handle_sendgrid_events():
    """Receive and process SendGrid event webhooks."""
    events = request.get_json()

    if not events:
        return jsonify({'error': 'No events'}), 400

    for event in events:
        event_type = event.get('event')
        email = event.get('email')
        timestamp = event.get('timestamp')
        sg_event_id = event.get('sg_event_id')
        sg_message_id = event.get('sg_message_id')

        print(f"[{event_type}] {email} at {timestamp}")

        if event_type == 'delivered':
            handle_delivered(email, event)
        elif event_type == 'open':
            handle_open(email, event)
        elif event_type == 'click':
            handle_click(email, event)
        elif event_type == 'bounce':
            handle_bounce(email, event)
        elif event_type == 'unsubscribe':
            handle_unsubscribe(email, event)
        elif event_type == 'dropped':
            handle_dropped(email, event)

    return jsonify({'status': 'ok'}), 200

def handle_delivered(email, event):
    """Email successfully delivered to recipient's server."""
    print(f"  ✅ Delivered to {email}")

def handle_open(email, event):
    """Recipient opened the email."""
    user_agent = event.get('useragent', 'unknown')
    ip = event.get('ip', 'unknown')
    print(f"  👁 Opened by {email} from {ip} ({user_agent})")

def handle_click(email, event):
    """Recipient clicked a link in the email."""
    url = event.get('url', 'unknown')
    print(f"  🖱 {email} clicked: {url}")

def handle_bounce(email, event):
    """Email bounced (invalid address or rejected by ISP)."""
    bounce_type = event.get('type', 'unknown')
    reason = event.get('reason', 'No reason')
    print(f"  ❌ Bounced: {email} ({bounce_type}): {reason}")
    # Remove from active list

def handle_unsubscribe(email, event):
    """Recipient unsubscribed."""
    print(f"  🚫 Unsubscribed: {email}")
    # Add to suppression list

def handle_dropped(email, event):
    """Email dropped before delivery (blocklist, invalid)."""
    reason = event.get('reason', 'unknown')
    print(f"  💧 Dropped: {email}: {reason}")

if __name__ == '__main__':
    # Configure in SendGrid: Settings → Mail Settings → Event Webhook
    # URL: https://myapp.com/sendgrid/events
    app.run(port=5000, debug=True)

# suppressions.py
# Manage bounces, blocks, and unsubscribes

from sendgrid import SendGridAPIClient
import os

sg = SendGridAPIClient(os.environ['SENDGRID_API_KEY'])

def list_bounces():
    """List all bounced email addresses."""
    response = sg.client.suppression.bounces.get()
    bounces = response.to_dict()
    print(f"Total bounces: {len(bounces)}")
    for bounce in bounces[:5]:
        print(f"  {bounce['email']} — {bounce.get('reason', 'No reason')[:50]}")
    return bounces

def delete_bounce(email):
    """Remove an email from the bounce list (after address is fixed)."""
    response = sg.client.suppression.bounces._(email).delete()
    print(f"Removed {email} from bounce list: {response.status_code}")

def list_suppressions():
    """List all suppression groups."""
    response = sg.client.asm.groups.get()
    groups = response.to_dict()
    print("Suppression Groups:")
    for group in groups:
        print(f"  {group['id']}: {group['name']} ({group['description'][:40]}...)")
    return groups

def add_to_suppression(email, group_id):
    """Add an email to a suppression group."""
    data = {"recipient_emails": [email]}
    response = sg.client.asm.groups._(group_id).suppressions.post(request_body=data)
    print(f"Added {email} to suppression group {group_id}: {response.status_code}")

def check_suppression(email):
    """Check if an email is in any suppression list."""
    response = sg.client.suppression.unsubscribes.get(query_params={"emails": email})
    result = response.to_dict()
    if result:
        print(f"⚠️ {email} is in suppression list")
    else:
        print(f"✅ {email} is not suppressed")
    return result

# Usage
# list_bounces()
# check_suppression('bounced@example.com')

# reputation.py
# Monitor sender reputation and deliverability stats

from sendgrid import SendGridAPIClient
from datetime import datetime, timedelta
import os

sg = SendGridAPIClient(os.environ['SENDGRID_API_KEY'])

def check_reputation():
    """Check sender reputation and recent stats."""
    now = datetime.now()
    start = now - timedelta(days=7)
    params = {
        'start_date': start.strftime('%Y-%m-%d'),
        'end_date': now.strftime('%Y-%m-%d'),
        'aggregated_by': 'day',
    }

    response = sg.client.stats.get(query_params=params)
    stats = response.to_dict()

    print("═" * 50)
    print("SendGrid Sender Reputation — Last 7 Days")
    print("═" * 50)

    totals = {'requests': 0, 'delivered': 0, 'opens': 0, 'clicks': 0,
              'bounces': 0, 'spam_reports': 0}

    for day in stats:
        for stat in day.get('stats', []):
            metrics = stat.get('metrics', {})
            totals['requests'] += metrics.get('requests', 0)
            totals['delivered'] += metrics.get('delivered', 0)
            totals['opens'] += metrics.get('unique_opens', 0)
            totals['clicks'] += metrics.get('unique_clicks', 0)
            totals['bounces'] += metrics.get('bounces', 0)
            totals['spam_reports'] += metrics.get('spam_reports', 0)

    delivery_rate = (totals['delivered'] / max(totals['requests'], 1)) * 100
    open_rate = (totals['opens'] / max(totals['delivered'], 1)) * 100
    bounce_rate = (totals['bounces'] / max(totals['requests'], 1)) * 100

    print(f"  Requests:      {totals['requests']:>8}")
    print(f"  Delivered:     {totals['delivered']:>8} ({delivery_rate:.1f}%)")
    print(f"  Unique Opens:  {totals['opens']:>8} ({open_rate:.1f}%)")
    print(f"  Unique Clicks: {totals['clicks']:>8}")
    print(f"  Bounces:       {totals['bounces']:>8} ({bounce_rate:.1f}%)")
    print(f"  Spam Reports:  {totals['spam_reports']:>8}")
    print("═" * 50)

    # Health assessment
    if bounce_rate > 5:
        print("⚠️  Warning: Bounce rate exceeds 5%. Clean your list.")
    if totals['spam_reports'] > 10:
        print("⚠️  Warning: High spam complaint rate. Review content.")

if __name__ == '__main__':
    check_reputation()

# smtp_send.py
# Send email via SendGrid SMTP (alternative to REST API)

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import os

def send_via_smtp(to_email, subject, html_body):
    """Send email via SendGrid SMTP relay."""
    # Create message
    msg = MIMEMultipart('alternative')
    msg['Subject'] = subject
    msg['From'] = 'noreply@dodatech.com'
    msg['To'] = to_email
    msg.attach(MIMEText(html_body, 'html'))

    # Send via SMTP
    try:
        server = smtplib.SMTP('smtp.sendgrid.net', 587)
        server.starttls()
        server.login('apikey', os.environ['SENDGRID_API_KEY'])  # Username is literally 'apikey'
        server.sendmail('noreply@dodatech.com', to_email, msg.as_string())
        server.quit()
        print(f"SMTP email sent to {to_email}")
    except Exception as e:
        print(f"SMTP failed: {e}")

send_via_smtp(
    to_email='alice@example.com',
    subject='Test from SMTP',
    html_body='<h1>Hello</h1><p>Sent via SMTP relay.</p>'
)

# email_dashboard.py
# Simple email analytics from SendGrid stats

from sendgrid import SendGridAPIClient
from datetime import datetime, timedelta
import os

sg = SendGridAPIClient(os.environ['SENDGRID_API_KEY'])

def email_analytics(days=30):
    """Track email performance over time."""
    now = datetime.now()
    start = now - timedelta(days=days)

    params = {
        'start_date': start.strftime('%Y-%m-%d'),
        'end_date': now.strftime('%Y-%m-%d'),
        'aggregated_by': 'week',
        'categories': 'transactional',
    }

    response = sg.client.stats.get(query_params=params)
    stats = response.to_dict()

    print("=" * 60)
    print(f"SendGrid Email Analytics — Last {days} Days")
    print("=" * 60)
    print(f"{'Week':<12} {'Sent':<8} {'Delivered':<10} {'Opens':<8} {'Clicks':<8} {'Bounce':<8}")
    print("-" * 60)

    for period in stats:
        date = period.get('date', 'unknown')
        metrics = period.get('stats', [{}])[0].get('metrics', {})

        sent = metrics.get('requests', 0)
        delivered = metrics.get('delivered', 0)
        opens = metrics.get('unique_opens', 0)
        clicks = metrics.get('unique_clicks', 0)
        bounces = metrics.get('bounces', 0)

        print(f"{date:<12} {sent:<8} {delivered:<10} {opens:<8} {clicks:<8} {bounces:<8}")

if __name__ == '__main__':
    email_analytics(days=30)