Python Automation Master Part 6: Email and Notification Automation
Mastering Multi-Channel Alert Systems with Python
Introduction: The Need for Notification Automation
The completeness of an automation system is determined by its notification capabilities. No matter how excellent your automation script is, it loses half its value if you can't check its results in a timely manner. Receiving appropriate notifications for various situations like server failures, batch job completions, and daily report deliveries greatly improves work efficiency.
In this Part 6, we'll learn how to send notifications through various channels using Python. We'll cover the most commonly used notification methods in practice, from traditional email to Slack, Telegram, and Discord.
1. Sending Email with smtplib
Python's built-in smtplib library allows you to send emails without any additional installation.
1.1 Basic Email Sending
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
def send_simple_email(sender, password, recipient, subject, body):
"""Send a simple text email"""
# Create message object
msg = MIMEMultipart()
msg['From'] = sender
msg['To'] = recipient
msg['Subject'] = subject
# Add body
msg.attach(MIMEText(body, 'plain', 'utf-8'))
try:
# Connect to Gmail SMTP server
server = smtplib.SMTP('smtp.gmail.com', 587)
server.starttls() # TLS encryption
server.login(sender, password)
# Send email
server.send_message(msg)
server.quit()
print("Email sent successfully!")
return True
except Exception as e:
print(f"Email sending failed: {e}")
return False
# Usage example
send_simple_email(
sender='your_email@gmail.com',
password='app_password', # Use app password after 2-step verification
recipient='recipient@example.com',
subject='[Automation] Daily Report',
body='Today\'s task has been completed successfully.'
)
2. Gmail/Outlook SMTP Settings
2.1 Gmail SMTP Settings
To use Gmail, you need to generate an app password first.
- Go to Google Account settings
- Enable 2-Step Verification in the Security tab
- Generate an App Password (select Other app)
- Use the generated 16-character password in your code
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
class GmailSender:
"""Email sending class using Gmail SMTP"""
def __init__(self, email, app_password):
self.email = email
self.password = app_password
self.smtp_server = 'smtp.gmail.com'
self.smtp_port = 587
def send(self, to, subject, body, html=False):
"""Send email"""
msg = MIMEMultipart('alternative')
msg['From'] = self.email
msg['To'] = to if isinstance(to, str) else ', '.join(to)
msg['Subject'] = subject
# Text or HTML body
content_type = 'html' if html else 'plain'
msg.attach(MIMEText(body, content_type, 'utf-8'))
with smtplib.SMTP(self.smtp_server, self.smtp_port) as server:
server.starttls()
server.login(self.email, self.password)
# Send to multiple recipients
recipients = [to] if isinstance(to, str) else to
server.send_message(msg)
return True
# Usage example
gmail = GmailSender('your_email@gmail.com', 'app_password')
gmail.send(
to='recipient@example.com',
subject='Test Email',
body='This is an email sent with Python.'
)
2.2 Outlook SMTP Settings
class OutlookSender:
"""Email sending class using Outlook SMTP"""
def __init__(self, email, password):
self.email = email
self.password = password
self.smtp_server = 'smtp-mail.outlook.com'
self.smtp_port = 587
def send(self, to, subject, body, html=False):
"""Send email"""
msg = MIMEMultipart('alternative')
msg['From'] = self.email
msg['To'] = to if isinstance(to, str) else ', '.join(to)
msg['Subject'] = subject
content_type = 'html' if html else 'plain'
msg.attach(MIMEText(body, content_type, 'utf-8'))
with smtplib.SMTP(self.smtp_server, self.smtp_port) as server:
server.starttls()
server.login(self.email, self.password)
server.send_message(msg)
return True
# Outlook mail settings:
# 1. Go to Outlook Security settings
# 2. Enable app-specific passwords if 2FA is enabled
# 3. Use actual Outlook account password or app password
outlook = OutlookSender('your_id@outlook.com', 'your_password')
outlook.send(
to='recipient@example.com',
subject='Outlook Mail Test',
body='This is an email sent with Outlook SMTP.'
)
3. Creating Email Body (Text, HTML)
3.1 HTML Format Email
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import smtplib
def send_html_email(sender, password, recipient, subject, data):
"""Send HTML formatted report email"""
# HTML template
html_template = """
<!DOCTYPE html>
<html>
<head>
<style>
body {{ font-family: Arial, sans-serif; }}
.container {{ max-width: 600px; margin: 0 auto; padding: 20px; }}
.header {{ background: #2c3e50; color: white; padding: 20px; text-align: center; }}
.content {{ padding: 20px; background: #f9f9f9; }}
.table {{ width: 100%; border-collapse: collapse; margin: 20px 0; }}
.table th, .table td {{ border: 1px solid #ddd; padding: 12px; text-align: left; }}
.table th {{ background: #3498db; color: white; }}
.table tr:nth-child(even) {{ background: #f2f2f2; }}
.footer {{ text-align: center; padding: 20px; color: #666; font-size: 12px; }}
.status-ok {{ color: #27ae60; font-weight: bold; }}
.status-error {{ color: #e74c3c; font-weight: bold; }}
</style>
</head>
<body>
<div class="container">
<div class="header">
<h1>{title}</h1>
<p>{date}</p>
</div>
<div class="content">
<h2>Summary</h2>
<p>Total processed: <strong>{total_count}</strong></p>
<p>Success: <span class="status-ok">{success_count}</span> / Failed: <span class="status-error">{fail_count}</span></p>
<h2>Details</h2>
<table class="table">
<tr>
<th>Item</th>
<th>Status</th>
<th>Processing Time</th>
</tr>
{table_rows}
</table>
</div>
<div class="footer">
<p>This email was sent automatically.</p>
<p>Contact: admin@example.com</p>
</div>
</div>
</body>
</html>
"""
# Generate table rows
table_rows = ""
for item in data['items']:
status_class = 'status-ok' if item['status'] == 'Success' else 'status-error'
table_rows += f"""
<tr>
<td>{item['name']}</td>
<td class="{status_class}">{item['status']}</td>
<td>{item['time']}</td>
</tr>
"""
# Complete HTML
html_content = html_template.format(
title=data['title'],
date=data['date'],
total_count=data['total_count'],
success_count=data['success_count'],
fail_count=data['fail_count'],
table_rows=table_rows
)
# Compose message
msg = MIMEMultipart('alternative')
msg['From'] = sender
msg['To'] = recipient
msg['Subject'] = subject
# Text version (for clients that don't support HTML)
text_content = f"""
{data['title']}
Date: {data['date']}
Total processed: {data['total_count']}
Success: {data['success_count']} / Failed: {data['fail_count']}
"""
msg.attach(MIMEText(text_content, 'plain', 'utf-8'))
msg.attach(MIMEText(html_content, 'html', 'utf-8'))
# Send
with smtplib.SMTP('smtp.gmail.com', 587) as server:
server.starttls()
server.login(sender, password)
server.send_message(msg)
return True
# Usage example
report_data = {
'title': 'Daily Batch Job Report',
'date': '2026-01-22',
'total_count': 5,
'success_count': 4,
'fail_count': 1,
'items': [
{'name': 'Data Collection', 'status': 'Success', 'time': '2.3s'},
{'name': 'Data Transformation', 'status': 'Success', 'time': '5.1s'},
{'name': 'Data Loading', 'status': 'Success', 'time': '3.7s'},
{'name': 'Backup Creation', 'status': 'Failed', 'time': '-'},
{'name': 'Notification', 'status': 'Success', 'time': '0.5s'}
]
}
send_html_email(
sender='your_email@gmail.com',
password='app_password',
recipient='recipient@example.com',
subject='[Report] Daily Batch Job Results',
data=report_data
)
4. Sending Attachments
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.base import MIMEBase
from email.mime.application import MIMEApplication
from email import encoders
import smtplib
import os
def send_email_with_attachments(sender, password, recipient, subject, body, attachments):
"""Send email with attachments
Args:
attachments: List of file paths to attach
"""
msg = MIMEMultipart()
msg['From'] = sender
msg['To'] = recipient
msg['Subject'] = subject
# Add body
msg.attach(MIMEText(body, 'plain', 'utf-8'))
# Add attachments
for file_path in attachments:
if not os.path.exists(file_path):
print(f"File not found: {file_path}")
continue
# Read file
with open(file_path, 'rb') as f:
file_data = f.read()
# Determine MIME type
file_name = os.path.basename(file_path)
part = MIMEApplication(file_data, Name=file_name)
# Add header
part['Content-Disposition'] = f'attachment; filename="{file_name}"'
msg.attach(part)
# Send
with smtplib.SMTP('smtp.gmail.com', 587) as server:
server.starttls()
server.login(sender, password)
server.send_message(msg)
print(f"Email sent ({len(attachments)} attachments)")
return True
# Usage example
send_email_with_attachments(
sender='your_email@gmail.com',
password='app_password',
recipient='recipient@example.com',
subject='Monthly Report Attached',
body='Hello,\n\nPlease find the attached monthly report.\n\nBest regards.',
attachments=[
'reports/monthly_report_202601.xlsx',
'reports/chart_202601.png'
]
)
5. Receiving and Parsing Emails (imaplib)
import imaplib
import email
from email.header import decode_header
from datetime import datetime
class EmailReader:
"""Email receiving and parsing class using IMAP"""
def __init__(self, email_addr, password, imap_server='imap.gmail.com'):
self.email = email_addr
self.password = password
self.imap_server = imap_server
self.mail = None
def connect(self):
"""Connect to IMAP server"""
self.mail = imaplib.IMAP4_SSL(self.imap_server)
self.mail.login(self.email, self.password)
return self
def disconnect(self):
"""Close connection"""
if self.mail:
self.mail.logout()
def get_folders(self):
"""Get folder list"""
status, folders = self.mail.list()
return [f.decode().split(' "/" ')[-1].strip('"') for f in folders]
def search_emails(self, folder='INBOX', criteria='ALL', limit=10):
"""Search emails
criteria examples:
- 'ALL': All emails
- 'UNSEEN': Unread emails
- 'FROM "sender@example.com"': From specific sender
- 'SUBJECT "keyword"': Subject contains keyword
- 'SINCE "01-Jan-2026"': Since specific date
"""
self.mail.select(folder)
status, messages = self.mail.search(None, criteria)
email_ids = messages[0].split()
# Sort by newest and apply limit
email_ids = email_ids[-limit:][::-1]
emails = []
for email_id in email_ids:
email_data = self._fetch_email(email_id)
if email_data:
emails.append(email_data)
return emails
def _fetch_email(self, email_id):
"""Fetch individual email"""
status, data = self.mail.fetch(email_id, '(RFC822)')
if status != 'OK':
return None
raw_email = data[0][1]
msg = email.message_from_bytes(raw_email)
# Decode headers
subject = self._decode_header(msg['Subject'])
from_addr = self._decode_header(msg['From'])
date_str = msg['Date']
# Extract body
body = self._get_body(msg)
# Attachment info
attachments = self._get_attachments(msg)
return {
'id': email_id.decode(),
'subject': subject,
'from': from_addr,
'date': date_str,
'body': body,
'attachments': attachments
}
def _decode_header(self, header):
"""Decode header"""
if header is None:
return ""
decoded_parts = decode_header(header)
result = []
for content, encoding in decoded_parts:
if isinstance(content, bytes):
content = content.decode(encoding or 'utf-8', errors='replace')
result.append(content)
return ''.join(result)
def _get_body(self, msg):
"""Extract body"""
body = ""
if msg.is_multipart():
for part in msg.walk():
content_type = part.get_content_type()
content_disposition = str(part.get('Content-Disposition'))
if content_type == 'text/plain' and 'attachment' not in content_disposition:
payload = part.get_payload(decode=True)
charset = part.get_content_charset() or 'utf-8'
body = payload.decode(charset, errors='replace')
break
else:
payload = msg.get_payload(decode=True)
charset = msg.get_content_charset() or 'utf-8'
body = payload.decode(charset, errors='replace')
return body
def _get_attachments(self, msg):
"""Extract attachment information"""
attachments = []
for part in msg.walk():
if part.get_content_maintype() == 'multipart':
continue
filename = part.get_filename()
if filename:
filename = self._decode_header(filename)
attachments.append({
'filename': filename,
'content_type': part.get_content_type(),
'size': len(part.get_payload(decode=True) or b'')
})
return attachments
# Usage example
reader = EmailReader('your_email@gmail.com', 'app_password')
reader.connect()
# Get unread emails
unread_emails = reader.search_emails(criteria='UNSEEN', limit=5)
for mail in unread_emails:
print(f"Subject: {mail['subject']}")
print(f"From: {mail['from']}")
print(f"Date: {mail['date']}")
print(f"Attachments: {len(mail['attachments'])}")
print("-" * 50)
reader.disconnect()
6. Sending Slack Notifications (Webhook)
Slack webhooks are very convenient as they can send messages with just HTTP requests without any special libraries.
6.1 Creating Webhook URL
- Go to Slack App page (api.slack.com/apps)
- Click Create New App
- Select From scratch, then set app name and workspace
- Enable Incoming Webhooks
- Click Add New Webhook to Workspace
- Select channel and copy webhook URL
import requests
import json
from datetime import datetime
class SlackNotifier:
"""Notification class using Slack webhooks"""
def __init__(self, webhook_url):
self.webhook_url = webhook_url
def send_simple(self, message):
"""Send simple text message"""
payload = {'text': message}
response = requests.post(
self.webhook_url,
data=json.dumps(payload),
headers={'Content-Type': 'application/json'}
)
return response.status_code == 200
def send_formatted(self, title, message, color='#36a64f', fields=None):
"""Send formatted message (using attachment)"""
attachment = {
'fallback': title,
'color': color,
'title': title,
'text': message,
'ts': datetime.now().timestamp()
}
if fields:
attachment['fields'] = [
{'title': k, 'value': str(v), 'short': True}
for k, v in fields.items()
]
payload = {'attachments': [attachment]}
response = requests.post(
self.webhook_url,
data=json.dumps(payload),
headers={'Content-Type': 'application/json'}
)
return response.status_code == 200
def send_alert(self, title, message, status='success'):
"""Send status-based alert"""
colors = {
'success': '#36a64f', # Green
'warning': '#ffcc00', # Yellow
'error': '#ff0000', # Red
'info': '#3498db' # Blue
}
emojis = {
'success': ':white_check_mark:',
'warning': ':warning:',
'error': ':x:',
'info': ':information_source:'
}
blocks = [
{
'type': 'header',
'text': {
'type': 'plain_text',
'text': f"{emojis.get(status, '')} {title}"
}
},
{
'type': 'section',
'text': {
'type': 'mrkdwn',
'text': message
}
},
{
'type': 'context',
'elements': [
{
'type': 'mrkdwn',
'text': f"Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
}
]
}
]
payload = {'blocks': blocks}
response = requests.post(
self.webhook_url,
data=json.dumps(payload),
headers={'Content-Type': 'application/json'}
)
return response.status_code == 200
# Usage example
slack = SlackNotifier('https://hooks.slack.com/services/xxx/yyy/zzz')
# Simple message
slack.send_simple('Batch job completed.')
# Formatted message
slack.send_formatted(
title='Daily Report',
message='Today\'s sales summary.',
color='#3498db',
fields={
'Total Orders': '150',
'Total Revenue': '$15,000',
'New Signups': '25',
'Refunds': '3'
}
)
# Error alert
slack.send_alert(
title='Server Error Detected',
message='*DB Connection Failed*\n```ConnectionError: Unable to connect to database```',
status='error'
)
7. Creating a Telegram Bot
7.1 Bot Creation and Setup
- Search for @BotFather on Telegram
- Enter /newbot command
- Set bot name and username
- Save the issued token
- Start a conversation with the created bot and get chat_id
import requests
from datetime import datetime
class TelegramBot:
"""Telegram bot notification class"""
def __init__(self, token):
self.token = token
self.base_url = f'https://api.telegram.org/bot{token}'
def get_updates(self):
"""Get recent messages (to find chat_id)"""
response = requests.get(f'{self.base_url}/getUpdates')
return response.json()
def send_message(self, chat_id, text, parse_mode='HTML'):
"""Send message
parse_mode: 'HTML' or 'Markdown'
"""
params = {
'chat_id': chat_id,
'text': text,
'parse_mode': parse_mode
}
response = requests.post(f'{self.base_url}/sendMessage', data=params)
return response.json()
def send_photo(self, chat_id, photo_path, caption=''):
"""Send image"""
with open(photo_path, 'rb') as photo:
params = {'chat_id': chat_id, 'caption': caption}
files = {'photo': photo}
response = requests.post(
f'{self.base_url}/sendPhoto',
data=params,
files=files
)
return response.json()
def send_document(self, chat_id, file_path, caption=''):
"""Send file"""
with open(file_path, 'rb') as doc:
params = {'chat_id': chat_id, 'caption': caption}
files = {'document': doc}
response = requests.post(
f'{self.base_url}/sendDocument',
data=params,
files=files
)
return response.json()
def send_alert(self, chat_id, title, message, status='info'):
"""Send formatted alert"""
emojis = {
'success': '✓',
'warning': '⚠',
'error': '✗',
'info': 'ℹ'
}
html_message = f"""
<b>{emojis.get(status, '')} {title}</b>
{message}
<i>Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</i>
"""
return self.send_message(chat_id, html_message)
def send_report(self, chat_id, title, data):
"""Send report format notification"""
lines = [f"<b>📊 {title}</b>\n"]
for key, value in data.items():
lines.append(f"• <b>{key}:</b> {value}")
lines.append(f"\n<i>Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}</i>")
return self.send_message(chat_id, '\n'.join(lines))
# Usage example
bot = TelegramBot('YOUR_BOT_TOKEN')
# Get chat_id (only once at first)
# Send a message to the bot then run this
# updates = bot.get_updates()
# print(updates)
chat_id = 'YOUR_CHAT_ID'
# Simple message
bot.send_message(chat_id, 'Hello! This is a Telegram bot.')
# Alert
bot.send_alert(
chat_id,
'Batch Job Completed',
'Daily data collection completed successfully.',
status='success'
)
# Report
bot.send_report(
chat_id,
'Daily Sales Summary',
{
'Total Orders': '150',
'Total Revenue': '$15,000',
'Average Order Value': '$100',
'New Customers': '25'
}
)
# File sending
bot.send_document(chat_id, 'reports/daily_report.xlsx', 'Daily report file')
8. Discord Webhooks
import requests
import json
from datetime import datetime
class DiscordNotifier:
"""Notification class using Discord webhooks"""
def __init__(self, webhook_url):
self.webhook_url = webhook_url
def send_simple(self, content):
"""Send simple message"""
payload = {'content': content}
response = requests.post(
self.webhook_url,
data=json.dumps(payload),
headers={'Content-Type': 'application/json'}
)
return response.status_code in [200, 204]
def send_embed(self, title, description, color=0x3498db, fields=None, footer=None):
"""Send embed message"""
embed = {
'title': title,
'description': description,
'color': color,
'timestamp': datetime.utcnow().isoformat()
}
if fields:
embed['fields'] = [
{'name': k, 'value': str(v), 'inline': True}
for k, v in fields.items()
]
if footer:
embed['footer'] = {'text': footer}
payload = {'embeds': [embed]}
response = requests.post(
self.webhook_url,
data=json.dumps(payload),
headers={'Content-Type': 'application/json'}
)
return response.status_code in [200, 204]
def send_alert(self, title, message, status='info'):
"""Send status-based alert"""
colors = {
'success': 0x2ecc71, # Green
'warning': 0xf39c12, # Orange
'error': 0xe74c3c, # Red
'info': 0x3498db # Blue
}
return self.send_embed(
title=title,
description=message,
color=colors.get(status, 0x3498db),
footer=f"Notification sent at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
)
# Usage example
discord = DiscordNotifier('https://discord.com/api/webhooks/xxx/yyy')
# Simple message
discord.send_simple('Batch job completed!')
# Embed message
discord.send_embed(
title='Daily Sales Report',
description='Today\'s sales summary',
color=0x3498db,
fields={
'Total Orders': '150',
'Revenue': '$15,000',
'New Customers': '25'
}
)
# Error alert
discord.send_alert(
title='Server Error',
message='Database connection failed. Please check immediately.',
status='error'
)
9. Unified Notification Manager
from datetime import datetime
class NotificationManager:
"""Unified notification manager for multiple channels"""
def __init__(self):
self.channels = {}
def add_email(self, name, sender, password, smtp_server='smtp.gmail.com', smtp_port=587):
"""Add email channel"""
self.channels[name] = {
'type': 'email',
'sender': sender,
'password': password,
'smtp_server': smtp_server,
'smtp_port': smtp_port
}
def add_slack(self, name, webhook_url):
"""Add Slack channel"""
self.channels[name] = {
'type': 'slack',
'webhook_url': webhook_url
}
def add_telegram(self, name, token, chat_id):
"""Add Telegram channel"""
self.channels[name] = {
'type': 'telegram',
'token': token,
'chat_id': chat_id
}
def add_discord(self, name, webhook_url):
"""Add Discord channel"""
self.channels[name] = {
'type': 'discord',
'webhook_url': webhook_url
}
def send(self, channel_name, title, message, **kwargs):
"""Send notification to specified channel"""
if channel_name not in self.channels:
raise ValueError(f"Channel not found: {channel_name}")
channel = self.channels[channel_name]
if channel['type'] == 'slack':
return self._send_slack(channel, title, message)
elif channel['type'] == 'telegram':
return self._send_telegram(channel, title, message)
elif channel['type'] == 'discord':
return self._send_discord(channel, title, message)
elif channel['type'] == 'email':
return self._send_email(channel, title, message, kwargs.get('recipient'))
def broadcast(self, title, message, **kwargs):
"""Send notification to all channels"""
results = {}
for name in self.channels:
try:
results[name] = self.send(name, title, message, **kwargs)
except Exception as e:
results[name] = f"Error: {e}"
return results
# Usage example
notifier = NotificationManager()
# Register channels
notifier.add_slack('dev-alerts', 'https://hooks.slack.com/services/xxx')
notifier.add_telegram('admin-bot', 'BOT_TOKEN', 'CHAT_ID')
notifier.add_discord('server-status', 'https://discord.com/api/webhooks/xxx')
# Send to specific channel
notifier.send('dev-alerts', 'Deployment Complete', 'v2.0.1 has been deployed successfully.')
# Broadcast to all channels
notifier.broadcast('Critical Alert', 'Server CPU usage exceeded 90%!')
Conclusion
In this article, we learned various notification automation methods using Python.
- Email (smtplib/imaplib): Ideal for formal reports and document delivery
- Slack Webhook: Best for team communication and real-time alerts
- Telegram Bot: Great for personal notifications and mobile alerts
- Discord Webhook: Perfect for community and server monitoring
In practice, you'll combine multiple channels according to the urgency and importance of notifications. For example, you might send daily reports via email while sending critical error alerts through Slack and Telegram simultaneously.
In the next Part 7, we'll cover API integration and data collection. You'll learn how to collect various data using REST APIs, including public data APIs, and integrate them with automation systems.