Email and Notification Automation: Sending Messages Programmatically

Automating email and notifications enables building responsive applications. This guide covers sending emails, SMS, and push notifications.

Email Automation

Sending Emails with SMTP

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders

# SMTP configuration
SMTP_SERVER = "smtp.gmail.com"
SMTP_PORT = 587
SENDER_EMAIL = "[email protected]"
SENDER_PASSWORD = "your_app_password"

def send_email(recipient, subject, body):
    """Send simple email"""
    try:
        # Create message
        message = MIMEText(body)
        message["Subject"] = subject
        message["From"] = SENDER_EMAIL
        message["To"] = recipient
        
        # Connect and send
        with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
            server.starttls()
            server.login(SENDER_EMAIL, SENDER_PASSWORD)
            server.send_message(message)
        
        print(f"Email sent to {recipient}")
    except Exception as e:
        print(f"Error sending email: {e}")

# Usage
send_email("[email protected]", "Hello", "This is a test email")

HTML Emails

def send_html_email(recipient, subject, html_body):
    """Send HTML email"""
    try:
        message = MIMEMultipart("alternative")
        message["Subject"] = subject
        message["From"] = SENDER_EMAIL
        message["To"] = recipient
        
        # Create HTML part
        html_part = MIMEText(html_body, "html")
        message.attach(html_part)
        
        # Send
        with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
            server.starttls()
            server.login(SENDER_EMAIL, SENDER_PASSWORD)
            server.send_message(message)
        
        print(f"HTML email sent to {recipient}")
    except Exception as e:
        print(f"Error: {e}")

# HTML template
html_template = """
<html>
  <body>
    <h1>Welcome!</h1>
    <p>Thank you for signing up.</p>
    <a href="https://example.com">Click here</a>
  </body>
</html>
"""

send_html_email("[email protected]", "Welcome", html_template)

Emails with Attachments

def send_email_with_attachment(recipient, subject, body, attachment_path):
    """Send email with attachment"""
    try:
        message = MIMEMultipart()
        message["Subject"] = subject
        message["From"] = SENDER_EMAIL
        message["To"] = recipient
        
        # Add body
        message.attach(MIMEText(body, "plain"))
        
        # Add attachment
        with open(attachment_path, "rb") as attachment:
            part = MIMEBase("application", "octet-stream")
            part.set_payload(attachment.read())
        
        encoders.encode_base64(part)
        part.add_header(
            "Content-Disposition",
            f"attachment; filename= {attachment_path}"
        )
        message.attach(part)
        
        # Send
        with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
            server.starttls()
            server.login(SENDER_EMAIL, SENDER_PASSWORD)
            server.send_message(message)
        
        print(f"Email with attachment sent to {recipient}")
    except Exception as e:
        print(f"Error: {e}")

send_email_with_attachment(
    "[email protected]",
    "Report",
    "Please find the report attached",
    "report.pdf"
)

Bulk Email Sending

def send_bulk_emails(recipients, subject, body):
    """Send emails to multiple recipients"""
    try:
        with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
            server.starttls()
            server.login(SENDER_EMAIL, SENDER_PASSWORD)
            
            for recipient in recipients:
                message = MIMEText(body)
                message["Subject"] = subject
                message["From"] = SENDER_EMAIL
                message["To"] = recipient
                
                server.send_message(message)
                print(f"Email sent to {recipient}")
    except Exception as e:
        print(f"Error: {e}")

# Usage
recipients = [
    "[email protected]",
    "[email protected]",
    "[email protected]"
]

send_bulk_emails(recipients, "Newsletter", "This month's newsletter...")

Using Email Services

SendGrid

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

def send_with_sendgrid(recipient, subject, body):
    """Send email using SendGrid"""
    message = Mail(
        from_email="[email protected]",
        to_emails=recipient,
        subject=subject,
        plain_text_content=body
    )
    
    try:
        sg = SendGridAPIClient("your_sendgrid_api_key")
        response = sg.send(message)
        print(f"Email sent: {response.status_code}")
    except Exception as e:
        print(f"Error: {e}")

send_with_sendgrid("[email protected]", "Hello", "Test email")

Mailgun

import requests

def send_with_mailgun(recipient, subject, body):
    """Send email using Mailgun"""
    domain = "mg.example.com"
    api_key = "your_mailgun_api_key"
    
    return requests.post(
        f"https://api.mailgun.net/v3/{domain}/messages",
        auth=("api", api_key),
        data={
            "from": "[email protected]",
            "to": recipient,
            "subject": subject,
            "text": body
        }
    )

send_with_mailgun("[email protected]", "Hello", "Test email")

SMS Notifications

Twilio SMS

from twilio.rest import Client

def send_sms(phone_number, message):
    """Send SMS using Twilio"""
    account_sid = "your_account_sid"
    auth_token = "your_auth_token"
    
    client = Client(account_sid, auth_token)
    
    try:
        message = client.messages.create(
            body=message,
            from_="+1234567890",  # Your Twilio number
            to=phone_number
        )
        print(f"SMS sent: {message.sid}")
    except Exception as e:
        print(f"Error: {e}")

send_sms("+1987654321", "Hello! This is a test SMS.")

AWS SNS

import boto3

def send_sms_with_sns(phone_number, message):
    """Send SMS using AWS SNS"""
    client = boto3.client('sns', region_name='us-east-1')
    
    try:
        response = client.publish(
            PhoneNumber=phone_number,
            Message=message
        )
        print(f"SMS sent: {response['MessageId']}")
    except Exception as e:
        print(f"Error: {e}")

send_sms_with_sns("+1987654321", "Hello from AWS SNS!")

Push Notifications

Firebase Cloud Messaging

from firebase_admin import credentials, messaging
import firebase_admin

# Initialize Firebase
cred = credentials.Certificate("serviceAccountKey.json")
firebase_admin.initialize_app(cred)

def send_push_notification(device_token, title, body):
    """Send push notification using FCM"""
    message = messaging.Message(
        notification=messaging.Notification(
            title=title,
            body=body
        ),
        token=device_token
    )
    
    try:
        response = messaging.send(message)
        print(f"Push notification sent: {response}")
    except Exception as e:
        print(f"Error: {e}")

send_push_notification(
    "device_token_here",
    "New Message",
    "You have a new message"
)

Send to Multiple Devices

def send_multicast_notification(device_tokens, title, body):
    """Send push notification to multiple devices"""
    message = messaging.MulticastMessage(
        notification=messaging.Notification(
            title=title,
            body=body
        ),
        tokens=device_tokens
    )
    
    try:
        response = messaging.send_multicast(message)
        print(f"Notifications sent: {response.success} successful")
    except Exception as e:
        print(f"Error: {e}")

device_tokens = ["token1", "token2", "token3"]
send_multicast_notification(
    device_tokens,
    "Announcement",
    "Important announcement"
)

Scheduled Notifications

Using APScheduler

from apscheduler.schedulers.background import BackgroundScheduler
import smtplib
from email.mime.text import MIMEText

scheduler = BackgroundScheduler()

def scheduled_email():
    """Send email on schedule"""
    message = MIMEText("Scheduled email body")
    message["Subject"] = "Scheduled Email"
    message["From"] = SENDER_EMAIL
    message["To"] = "[email protected]"
    
    with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
        server.starttls()
        server.login(SENDER_EMAIL, SENDER_PASSWORD)
        server.send_message(message)
    
    print("Scheduled email sent")

# Schedule email
scheduler.add_job(scheduled_email, 'cron', hour=9, minute=0)
scheduler.start()

# Keep running
try:
    while True:
        pass
except KeyboardInterrupt:
    scheduler.shutdown()

Using Celery

from celery import Celery
import smtplib
from email.mime.text import MIMEText

app = Celery('tasks', broker='redis://localhost:6379')

@app.task
def send_email_task(recipient, subject, body):
    """Send email as Celery task"""
    message = MIMEText(body)
    message["Subject"] = subject
    message["From"] = SENDER_EMAIL
    message["To"] = recipient
    
    with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
        server.starttls()
        server.login(SENDER_EMAIL, SENDER_PASSWORD)
        server.send_message(message)
    
    return f"Email sent to {recipient}"

# Send email asynchronously
send_email_task.delay("[email protected]", "Hello", "Test email")

# Schedule email
from celery.schedules import crontab

app.conf.beat_schedule = {
    'send-email-every-morning': {
        'task': 'tasks.send_email_task',
        'schedule': crontab(hour=9, minute=0),
        'args': ("[email protected]", "Morning Email", "Good morning!")
    }
}

Best Practices

Use environment variables: Store credentials securely
Error handling: Handle failures gracefully
Rate limiting: Respect service limits
Logging: Log all notifications
Templates: Use email templates for consistency
Unsubscribe: Provide unsubscribe option
Testing: Test with sandbox/test accounts

Common Pitfalls

Bad Practice:

# Don't: Hardcode credentials
SENDER_PASSWORD = "mypassword123"

# Don't: No error handling
server.send_message(message)

# Don't: Send without delay
for user in users:
    send_email(user)  # May get rate limited

Good Practice:

# Do: Use environment variables
import os
SENDER_PASSWORD = os.getenv("EMAIL_PASSWORD")

# Do: Handle errors
try:
    server.send_message(message)
except Exception as e:
    logger.error(f"Failed to send email: {e}")

# Do: Use queue/delay
for user in users:
    send_email_task.delay(user)  # Async with Celery

Conclusion

Email and notification automation enables building responsive applications. Choose appropriate services based on volume and requirements. Always handle errors, secure credentials, and respect rate limits.