Chapter 10
Email Automation — Batch Sending and Smart Inbox Processing
Chapter 10: Email Automation — Scheduled Reports, Alerts & Notifications
Email is the most universal asynchronous communication channel. Whether you're sending daily data reports, exception alerts, or personalized client updates, Python can handle it fully automatically. This chapter covers three email solutions (from stdlib to enterprise-grade), HTML template rendering, attachments, inbox parsing, and ends with a daily automated report system.
Python Email Library Comparison
| Library | Position | Strengths | Best for |
|---|---|---|---|
| smtplib (stdlib) | Built-in, no install | Zero dependencies, full control | Learning, custom needs |
| yagmail | Gmail-optimized wrapper | 3 lines to send email, auto attachments | Personal scripts, rapid prototyping |
| SendGrid SDK | Enterprise email service | High deliverability, analytics, templates | Production, marketing emails |
Sending Email Basics
SMTP Configuration
| Provider | SMTP Host | Port | Notes |
|---|---|---|---|
| Gmail | smtp.gmail.com | 587 | Requires App Password (not login password) |
| Outlook | smtp-mail.outlook.com | 587 | Regular credentials work |
| Office 365 | smtp.office365.com | 587 | May need admin to enable SMTP AUTH |
import smtplib, os
from email.message import EmailMessage
def send_text_email(subject: str, body: str, to: str | list[str]):
msg = EmailMessage()
msg["Subject"] = subject
msg["From"] = os.getenv("EMAIL_USER")
msg["To"] = to if isinstance(to, str) else ", ".join(to)
msg.set_content(body)
with smtplib.SMTP("smtp.gmail.com", 587) as server:
server.starttls()
server.login(os.getenv("EMAIL_USER"), os.getenv("EMAIL_PASS"))
server.send_message(msg)
Email with Attachments
from pathlib import Path
from email.message import EmailMessage
import smtplib, os
def send_with_attachment(subject: str, body: str, to: str, file_path: str):
msg = EmailMessage()
msg["Subject"] = subject
msg["From"] = os.getenv("EMAIL_USER")
msg["To"] = to
msg.set_content(body)
path = Path(file_path)
with open(path, "rb") as f:
msg.add_attachment(f.read(), maintype="application",
subtype="octet-stream", filename=path.name)
with smtplib.SMTP("smtp.gmail.com", 587) as server:
server.starttls()
server.login(os.getenv("EMAIL_USER"), os.getenv("EMAIL_PASS"))
server.send_message(msg)
HTML Email Templates with Jinja2
from jinja2 import Template
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import smtplib, os
REPORT_TEMPLATE = """
<html><body style="font-family:Arial,sans-serif;max-width:600px;margin:0 auto;">
<h2 style="color:#2563eb;">Weekly Report: {{"{{"}}week{{"}}"}}</h2>
<p>Revenue: <strong>${{"{{"}}revenue{{"}}"}}</strong>
<span style="color:{{"{{"}}'green' if change >= 0 else 'red'{{"}}"}}">({{"{{"}}'%+.1f' % change{{"}}"}}%)</span></p>
<table style="width:100%;border-collapse:collapse;">
<tr style="background:#2563eb;color:white;"><th style="padding:8px;">Product</th><th>Sales</th></tr>
{% for p in products %}
<tr><td style="padding:8px;border-bottom:1px solid #e5e7eb;">{{"{{"}}p.name{{"}}"}}</td>
<td style="padding:8px;border-bottom:1px solid #e5e7eb;">{{"{{"}}p.sales{{"}}"}}</td></tr>
{% endfor %}
</table>
</body></html>
"""
def send_html_email(subject: str, html: str, to: str):
msg = MIMEMultipart("alternative")
msg["Subject"] = subject
msg["From"] = os.getenv("EMAIL_USER")
msg["To"] = to
msg.attach(MIMEText("Please view in an HTML-capable email client.", "plain"))
msg.attach(MIMEText(html, "html", "utf-8"))
with smtplib.SMTP("smtp.gmail.com", 587) as server:
server.starttls()
server.login(os.getenv("EMAIL_USER"), os.getenv("EMAIL_PASS"))
server.sendmail(os.getenv("EMAIL_USER"), to, msg.as_string())
# Usage
html = Template(REPORT_TEMPLATE).render(
week="Week 3, 2024", revenue="128,000", change=12.5,
products=[{"name": "Headphones Pro", "sales": 320}]
)
send_html_email("Weekly Report — Week 3", html, "[email protected]")
Receiving and Parsing Email
import imaplib, email, os
from email.header import decode_header
from pathlib import Path
def fetch_and_save_attachments(save_dir: str = "attachments") -> list[str]:
"""Fetch unread emails and save all attachments."""
Path(save_dir).mkdir(exist_ok=True)
saved = []
with imaplib.IMAP4_SSL("imap.gmail.com") as imap:
imap.login(os.getenv("EMAIL_USER"), os.getenv("EMAIL_PASS"))
imap.select("INBOX")
_, ids = imap.search(None, "UNSEEN")
for mid in ids[0].split():
_, data = imap.fetch(mid, "(RFC822)")
msg = email.message_from_bytes(data[0][1])
for part in msg.walk():
if part.get_content_disposition() == "attachment":
fname, enc = decode_header(part.get_filename())[0]
if isinstance(fname, bytes):
fname = fname.decode(enc or "utf-8")
path = Path(save_dir) / fname
with open(path, "wb") as f:
f.write(part.get_payload(decode=True))
saved.append(str(path))
return saved
Enterprise Email: Bulk Personalized Sending
import csv, time
TEMPLATE = """Dear {name},
Your invoice #{invoice_id} for ${amount} is due on {due_date}.
Please arrange payment before the due date.
Best regards,
Finance Team"""
def send_bulk_personalized(csv_path: str):
with open(csv_path, encoding="utf-8") as f:
for i, row in enumerate(csv.DictReader(f)):
send_text_email(
subject=f"Invoice #{row['invoice_id']} — Payment Due",
body=TEMPLATE.format(**row),
to=row["email"],
)
print(f"[{i+1}] Sent to {row['name']} ({row['email']})")
time.sleep(2) # Rate limiting
Project: Automated Daily Report System
import os, sqlite3, smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from datetime import datetime, timedelta
from jinja2 import Template
RECIPIENTS = ["[email protected]", "[email protected]"]
TEMPLATE = """<html><body style="font-family:Arial,sans-serif;">
<h2>Daily Report: {{"{{"}}yesterday{{"}}"}}</h2>
<ul>
<li>Orders: <strong>{{"{{"}}orders{{"}}"}}</strong></li>
<li>Revenue: <strong>${{"{{"}}revenue{{"}}"}}</strong></li>
<li>New Users: <strong>{{"{{"}}new_users{{"}}"}}</strong></li>
</ul>
</body></html>"""
def get_stats(date_str: str) -> dict:
conn = sqlite3.connect("sales.db")
row = conn.execute(
"SELECT COUNT(*), COALESCE(SUM(amount),0) FROM orders WHERE date(created_at)=?",
(date_str,)
).fetchone()
users = conn.execute(
"SELECT COUNT(*) FROM users WHERE date(created_at)=?", (date_str,)
).fetchone()[0]
conn.close()
return {"orders": row[0], "revenue": f"{row[1]:,.0f}", "new_users": users}
def send_daily_report():
yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
stats = get_stats(yesterday)
stats["yesterday"] = yesterday
html = Template(TEMPLATE).render(**stats)
msg = MIMEMultipart("alternative")
msg["Subject"] = f"Daily Report — {yesterday}"
msg["From"] = os.getenv("EMAIL_USER")
msg["To"] = ", ".join(RECIPIENTS)
msg.attach(MIMEText(html, "html", "utf-8"))
with smtplib.SMTP("smtp.gmail.com", 587) as s:
s.starttls()
s.login(os.getenv("EMAIL_USER"), os.getenv("EMAIL_PASS"))
s.sendmail(os.getenv("EMAIL_USER"), RECIPIENTS, msg.as_string())
print(f"Daily report sent for {yesterday}")
if __name__ == "__main__":
send_daily_report()
Previous
Next
Chapter 11: Messaging Notifications