How to Send Emails in Flask (2026 Guide)

Most "how to send email in Flask" tutorials start with Flask-Mail and SMTP. That's fine for internal tools. It's not fine when you need to send welcome emails, password resets, payment receipts, and onboarding sequences to real users who expect emails to actually arrive.
This guide covers the full picture: picking a provider, building HTML email templates with Jinja2, sending from routes and form submissions, offloading to background threads and Celery, handling Stripe webhooks, and shipping to production. For a broader Python overview or framework-specific guides, see sending emails in Python, Django, or FastAPI. All code examples use Flask with Python type hints.
Pick an Email Provider
You have three solid options. Each code example below lets you switch between them.
- Sequenzy is built for SaaS. Transactional emails, marketing campaigns, automated sequences, subscriber management, all from one API. If you're building a SaaS product, this is the simplest path because you won't need to glue together three different tools later. Native Stripe integration and built-in retries.
- Resend is a developer-friendly transactional email API. Clean DX, good docs, solid deliverability. They have one-off broadcast campaigns but no automations or sequences.
- SendGrid is the enterprise standard. Feature-rich, sometimes complex. Good if you need high volume and don't mind a bigger API surface.
Install
pip install flask requests python-dotenvpip install flask requests python-dotenvpip install flask requests python-dotenvWe're using requests directly for all three providers instead of their Python SDKs. Every email provider is just an HTTP API. Using requests keeps your code consistent and gives you full control over error handling.
Add your API key to .env:
SEQUENZY_API_KEY=sq_your_api_key_hereRESEND_API_KEY=re_your_api_key_hereSENDGRID_API_KEY=SG.your_api_key_hereConfigure Your App
Use Flask's config system with python-dotenv to load environment variables:
# config.py
import os
from dotenv import load_dotenv
load_dotenv()
class Config:
SECRET_KEY = os.environ.get("SECRET_KEY", "dev-secret-key")
SEQUENZY_API_KEY = os.environ["SEQUENZY_API_KEY"]
APP_URL = os.environ.get("APP_URL", "https://yourapp.com")
FROM_EMAIL = os.environ.get("FROM_EMAIL", "Your App <noreply@yourdomain.com>")Create the Email Service
Create a reusable email service. Flask is synchronous, so we use requests with a Session for connection pooling:
import requests
import logging
logger = logging.getLogger(__name__)
class EmailService:
def __init__(self, api_key: str):
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
})
self.base_url = "https://api.sequenzy.com/v1"
def send(self, to: str, subject: str, body: str) -> dict:
response = self.session.post(
f"{self.base_url}/transactional/send",
json={"to": to, "subject": subject, "body": body},
timeout=30,
)
response.raise_for_status()
logger.info(f"Email sent to {to}: {subject}")
return response.json()import requests
import logging
logger = logging.getLogger(__name__)
class EmailService:
def __init__(self, api_key: str, from_email: str):
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
})
self.base_url = "https://api.resend.com"
self.from_email = from_email
def send(self, to: str, subject: str, html: str) -> dict:
response = self.session.post(
f"{self.base_url}/emails",
json={
"from": self.from_email,
"to": to,
"subject": subject,
"html": html,
},
timeout=30,
)
response.raise_for_status()
logger.info(f"Email sent to {to}: {subject}")
return response.json()import requests
import logging
logger = logging.getLogger(__name__)
class EmailService:
def __init__(self, api_key: str, from_email: str):
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
})
self.base_url = "https://api.sendgrid.com/v3"
self.from_email = from_email
def send(self, to: str, subject: str, html: str) -> dict:
response = self.session.post(
f"{self.base_url}/mail/send",
json={
"personalizations": [{"to": [{"email": to}]}],
"from": {"email": self.from_email},
"subject": subject,
"content": [{"type": "text/html", "value": html}],
},
timeout=30,
)
response.raise_for_status()
logger.info(f"Email sent to {to}: {subject}")
return {"sent": True}The requests.Session reuses TCP connections across requests. This is Flask's equivalent of connection pooling — much faster than creating a new connection per email.
Wire Up the App
Initialize the email service as a Flask extension pattern:
from flask import Flask
from config import Config
from email_service import EmailService
app = Flask(__name__)
app.config.from_object(Config)
email = EmailService(api_key=app.config["SEQUENZY_API_KEY"])from flask import Flask
from config import Config
from email_service import EmailService
app = Flask(__name__)
app.config.from_object(Config)
email = EmailService(
api_key=app.config["RESEND_API_KEY"],
from_email=app.config["FROM_EMAIL"],
)from flask import Flask
from config import Config
from email_service import EmailService
app = Flask(__name__)
app.config.from_object(Config)
email = EmailService(
api_key=app.config["SENDGRID_API_KEY"],
from_email=app.config["FROM_EMAIL"],
)Send Your First Email
The simplest possible email from a Flask route:
from flask import request, jsonify
@app.post("/api/send")
def send_welcome():
data = request.get_json()
name = data.get("name")
to = data.get("email")
if not name or not to:
return jsonify({"error": "name and email required"}), 400
result = email.send(
to=to,
subject=f"Welcome, {name}",
body=f"<h1>Welcome, {name}</h1><p>Your account is ready.</p>",
)
return jsonify({"jobId": result["jobId"]})from flask import request, jsonify
@app.post("/api/send")
def send_welcome():
data = request.get_json()
name = data.get("name")
to = data.get("email")
if not name or not to:
return jsonify({"error": "name and email required"}), 400
result = email.send(
to=to,
subject=f"Welcome, {name}",
html=f"<h1>Welcome, {name}</h1><p>Your account is ready.</p>",
)
return jsonify({"id": result["id"]})from flask import request, jsonify
@app.post("/api/send")
def send_welcome():
data = request.get_json()
name = data.get("name")
to = data.get("email")
if not name or not to:
return jsonify({"error": "name and email required"}), 400
email.send(
to=to,
subject=f"Welcome, {name}",
html=f"<h1>Welcome, {name}</h1><p>Your account is ready.</p>",
)
return jsonify({"sent": True})Test it:
flask run
curl -X POST http://localhost:5000/api/send \
-H "Content-Type: application/json" \
-d '{"email": "user@example.com", "name": "Alice"}'Build Email Templates with Jinja2
Flask ships with Jinja2. Use it for email templates the same way you use it for page templates:
project/
├── templates/
│ ├── emails/
│ │ ├── base.html
│ │ ├── welcome.html
│ │ ├── password-reset.html
│ │ └── receipt.html
│ └── pages/
│ └── contact.html
├── email_service.py
├── config.py
└── app.py
Start with a base email template:
<!-- templates/emails/base.html -->
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
</head>
<body style="margin:0;padding:0;background-color:#f6f9fc;font-family:sans-serif;">
<div style="max-width:480px;margin:40px auto;background:#fff;padding:40px;border-radius:8px;">
{% block content %}{% endblock %}
<hr style="border:none;border-top:1px solid #e5e7eb;margin:32px 0 16px;" />
<p style="font-size:12px;color:#9ca3af;">
© {{ current_year }} Your App. All rights reserved.
</p>
</div>
</body>
</html>Extend it for each email type:
<!-- templates/emails/welcome.html -->
{% extends "emails/base.html" %}
{% block content %}
<h1 style="font-size:24px;margin-bottom:16px;">Welcome, {{ name }}</h1>
<p style="font-size:16px;line-height:1.6;color:#374151;">
Your account is ready. You can log in and start exploring.
</p>
<a href="{{ login_url }}"
style="display:inline-block;background:#f97316;color:#fff;padding:12px 24px;border-radius:6px;text-decoration:none;font-size:14px;font-weight:600;margin-top:16px;">
Go to Dashboard
</a>
{% endblock %}Inject the current year automatically using a context processor:
# app.py
from datetime import datetime
@app.context_processor
def inject_globals():
return {"current_year": datetime.now().year}Render and send:
from flask import render_template
@app.post("/api/send-welcome")
def send_welcome():
data = request.get_json()
name = data["name"]
to = data["email"]
html = render_template("emails/welcome.html", name=name, login_url="https://app.yoursite.com")
result = email.send(to=to, subject=f"Welcome, {name}", body=html)
return jsonify({"jobId": result["jobId"]})from flask import render_template
@app.post("/api/send-welcome")
def send_welcome():
data = request.get_json()
name = data["name"]
to = data["email"]
html = render_template("emails/welcome.html", name=name, login_url="https://app.yoursite.com")
result = email.send(to=to, subject=f"Welcome, {name}", html=html)
return jsonify({"id": result["id"]})from flask import render_template
@app.post("/api/send-welcome")
def send_welcome():
data = request.get_json()
name = data["name"]
to = data["email"]
html = render_template("emails/welcome.html", name=name, login_url="https://app.yoursite.com")
email.send(to=to, subject=f"Welcome, {name}", html=html)
return jsonify({"sent": True})Flask's render_template works outside of HTTP requests too (you need the app context), which makes it great for background email tasks.
Form Submissions
Flask handles form-based email sends with request.form. Here's a contact form that sends an email and shows a success message:
from flask import redirect, url_for, flash
@app.post("/contact")
def contact():
sender_email = request.form.get("email", "").strip()
message = request.form.get("message", "").strip()
if not sender_email or not message:
flash("All fields are required.", "error")
return redirect(url_for("contact_page"))
try:
email.send(
to="you@yourcompany.com",
subject=f"Contact from {sender_email}",
body=f"<p><strong>From:</strong> {sender_email}</p><p>{message}</p>",
)
flash("Message sent! We'll get back to you soon.", "success")
except Exception:
flash("Something went wrong. Please try again.", "error")
return redirect(url_for("contact_page"))
@app.get("/contact")
def contact_page():
return render_template("pages/contact.html")from flask import redirect, url_for, flash
@app.post("/contact")
def contact():
sender_email = request.form.get("email", "").strip()
message = request.form.get("message", "").strip()
if not sender_email or not message:
flash("All fields are required.", "error")
return redirect(url_for("contact_page"))
try:
email.send(
to="you@yourcompany.com",
subject=f"Contact from {sender_email}",
html=f"<p><strong>From:</strong> {sender_email}</p><p>{message}</p>",
)
flash("Message sent! We'll get back to you soon.", "success")
except Exception:
flash("Something went wrong. Please try again.", "error")
return redirect(url_for("contact_page"))
@app.get("/contact")
def contact_page():
return render_template("pages/contact.html")from flask import redirect, url_for, flash
@app.post("/contact")
def contact():
sender_email = request.form.get("email", "").strip()
message = request.form.get("message", "").strip()
if not sender_email or not message:
flash("All fields are required.", "error")
return redirect(url_for("contact_page"))
try:
email.send(
to="you@yourcompany.com",
subject=f"Contact from {sender_email}",
html=f"<p><strong>From:</strong> {sender_email}</p><p>{message}</p>",
)
flash("Message sent! We'll get back to you soon.", "success")
except Exception:
flash("Something went wrong. Please try again.", "error")
return redirect(url_for("contact_page"))
@app.get("/contact")
def contact_page():
return render_template("pages/contact.html")The Post/Redirect/Get pattern prevents duplicate form submissions on page refresh.
Background Sending with Threads
Flask is synchronous, so email sends block the response. For non-critical emails (welcome emails, notifications), use a background thread so the user doesn't wait:
# background.py
import threading
import logging
from flask import Flask
logger = logging.getLogger(__name__)
def send_in_background(app: Flask, fn, *args, **kwargs):
"""Run a function in a background thread with app context."""
def _run():
with app.app_context():
try:
fn(*args, **kwargs)
except Exception:
logger.exception("Background task failed")
thread = threading.Thread(target=_run)
thread.start()The app.app_context() is important — without it, Flask functions like render_template won't work in the background thread.
Use it in your routes:
from background import send_in_background
from flask import current_app
def send_welcome_email(to: str, name: str):
html = render_template("emails/welcome.html", name=name, login_url="https://app.yoursite.com")
email.send(to=to, subject=f"Welcome, {name}", body=html)
@app.post("/api/signup")
def signup():
data = request.get_json()
name = data["name"]
to = data["email"]
# Create user in database first...
send_in_background(current_app._get_current_object(), send_welcome_email, to, name)
return jsonify({"message": "Account created"})from background import send_in_background
from flask import current_app
def send_welcome_email(to: str, name: str):
html = render_template("emails/welcome.html", name=name, login_url="https://app.yoursite.com")
email.send(to=to, subject=f"Welcome, {name}", html=html)
@app.post("/api/signup")
def signup():
data = request.get_json()
name = data["name"]
to = data["email"]
# Create user in database first...
send_in_background(current_app._get_current_object(), send_welcome_email, to, name)
return jsonify({"message": "Account created"})from background import send_in_background
from flask import current_app
def send_welcome_email(to: str, name: str):
html = render_template("emails/welcome.html", name=name, login_url="https://app.yoursite.com")
email.send(to=to, subject=f"Welcome, {name}", html=html)
@app.post("/api/signup")
def signup():
data = request.get_json()
name = data["name"]
to = data["email"]
# Create user in database first...
send_in_background(current_app._get_current_object(), send_welcome_email, to, name)
return jsonify({"message": "Account created"})We use current_app._get_current_object() instead of current_app directly because the proxy object doesn't work across threads.
Background Sending with Celery
For production apps with higher email volume, use Celery. It gives you persistent task queues, retries, and dedicated worker processes:
pip install celery redisfrom celery import Celery
from email_service import EmailService
from config import Config
celery = Celery("tasks", broker="redis://localhost:6379/0")
# Create a standalone email client for the worker process
email = EmailService(api_key=Config.SEQUENZY_API_KEY)
@celery.task(bind=True, max_retries=3, default_retry_delay=60)
def send_email_task(self, to: str, subject: str, body: str):
try:
email.send(to=to, subject=subject, body=body)
except Exception as exc:
self.retry(exc=exc)from celery import Celery
from email_service import EmailService
from config import Config
celery = Celery("tasks", broker="redis://localhost:6379/0")
email = EmailService(api_key=Config.RESEND_API_KEY, from_email=Config.FROM_EMAIL)
@celery.task(bind=True, max_retries=3, default_retry_delay=60)
def send_email_task(self, to: str, subject: str, html: str):
try:
email.send(to=to, subject=subject, html=html)
except Exception as exc:
self.retry(exc=exc)from celery import Celery
from email_service import EmailService
from config import Config
celery = Celery("tasks", broker="redis://localhost:6379/0")
email = EmailService(api_key=Config.SENDGRID_API_KEY, from_email=Config.FROM_EMAIL)
@celery.task(bind=True, max_retries=3, default_retry_delay=60)
def send_email_task(self, to: str, subject: str, html: str):
try:
email.send(to=to, subject=subject, html=html)
except Exception as exc:
self.retry(exc=exc)Use it in your routes:
from tasks import send_email_task
@app.post("/api/signup")
def signup():
data = request.get_json()
name = data["name"]
# Render the template in the request context
html = render_template("emails/welcome.html", name=name, login_url="https://app.yoursite.com")
# Queue the email — Celery handles retries
send_email_task.delay(to=data["email"], subject=f"Welcome, {name}", body=html)
return jsonify({"message": "Account created"})Start the worker:
celery -A tasks worker --loglevel=infoOrganize with Blueprints
Once your app grows, use Flask blueprints to organize email-related routes:
# blueprints/emails.py
from flask import Blueprint, request, jsonify, render_template, current_app
emails_bp = Blueprint("emails", __name__, url_prefix="/api")
@emails_bp.post("/send-welcome")
def send_welcome():
data = request.get_json()
name = data["name"]
to = data["email"]
html = render_template("emails/welcome.html", name=name, login_url="https://app.yoursite.com")
from app import email
result = email.send(to=to, subject=f"Welcome, {name}", body=html)
return jsonify(result)# app.py
from blueprints.emails import emails_bp
app.register_blueprint(emails_bp)Common Email Patterns for SaaS
Here are the emails almost every SaaS app needs.
Password Reset
from flask import render_template
def send_password_reset(email_service, to: str, reset_token: str):
reset_url = f"https://yourapp.com/reset-password?token={reset_token}"
html = render_template("emails/password-reset.html", reset_url=reset_url)
email_service.send(
to=to,
subject="Reset your password",
body=html,
)from flask import render_template
def send_password_reset(email_service, to: str, reset_token: str):
reset_url = f"https://yourapp.com/reset-password?token={reset_token}"
html = render_template("emails/password-reset.html", reset_url=reset_url)
email_service.send(
to=to,
subject="Reset your password",
html=html,
)from flask import render_template
def send_password_reset(email_service, to: str, reset_token: str):
reset_url = f"https://yourapp.com/reset-password?token={reset_token}"
html = render_template("emails/password-reset.html", reset_url=reset_url)
email_service.send(
to=to,
subject="Reset your password",
html=html,
)<!-- templates/emails/password-reset.html -->
{% extends "emails/base.html" %}
{% block content %}
<h2 style="font-size:20px;margin-bottom:16px;">Password Reset</h2>
<p style="font-size:16px;line-height:1.6;color:#374151;">
Click the button below to reset your password. This link expires in 1 hour.
</p>
<a href="{{ reset_url }}"
style="display:inline-block;background:#f97316;color:#fff;padding:12px 24px;border-radius:6px;text-decoration:none;font-weight:600;">
Reset Password
</a>
<p style="color:#6b7280;font-size:14px;margin-top:24px;">
If you didn't request this, ignore this email.
</p>
{% endblock %}Payment Receipt
from flask import render_template
def send_receipt(email_service, to: str, amount: int, plan: str, invoice_url: str):
formatted = f"${amount / 100:.2f}"
html = render_template(
"emails/receipt.html",
amount=formatted,
plan=plan,
invoice_url=invoice_url,
)
email_service.send(
to=to,
subject=f"Payment receipt - {formatted}",
body=html,
)from flask import render_template
def send_receipt(email_service, to: str, amount: int, plan: str, invoice_url: str):
formatted = f"${amount / 100:.2f}"
html = render_template(
"emails/receipt.html",
amount=formatted,
plan=plan,
invoice_url=invoice_url,
)
email_service.send(
to=to,
subject=f"Payment receipt - {formatted}",
html=html,
)from flask import render_template
def send_receipt(email_service, to: str, amount: int, plan: str, invoice_url: str):
formatted = f"${amount / 100:.2f}"
html = render_template(
"emails/receipt.html",
amount=formatted,
plan=plan,
invoice_url=invoice_url,
)
email_service.send(
to=to,
subject=f"Payment receipt - {formatted}",
html=html,
)Stripe Webhook Handler
import hmac
import hashlib
import time
from flask import Blueprint, request, jsonify, current_app
from emails.receipt import send_receipt
stripe_bp = Blueprint("stripe", __name__)
def verify_stripe_signature(payload: bytes, signature: str, secret: str) -> bool:
parts = dict(item.split("=", 1) for item in signature.split(","))
timestamp = parts.get("t", "")
expected_sig = parts.get("v1", "")
if not timestamp or not expected_sig:
return False
if abs(time.time() - int(timestamp)) > 300:
return False
signed_payload = f"{timestamp}.{payload.decode()}"
computed = hmac.new(
secret.encode(),
signed_payload.encode(),
hashlib.sha256,
).hexdigest()
return hmac.compare_digest(computed, expected_sig)
@stripe_bp.post("/api/webhooks/stripe")
def stripe_webhook():
payload = request.get_data()
signature = request.headers.get("Stripe-Signature", "")
secret = current_app.config["STRIPE_WEBHOOK_SECRET"]
if not verify_stripe_signature(payload, signature, secret):
return jsonify({"error": "Invalid signature"}), 400
event = request.get_json()
if event["type"] == "checkout.session.completed":
session = event["data"]["object"]
from app import email
send_receipt(
email,
to=session["customer_email"],
amount=session["amount_total"],
plan=session.get("metadata", {}).get("plan", "Pro"),
invoice_url=session.get("invoice_url", "https://yourapp.com/billing"),
)
if event["type"] == "invoice.payment_failed":
invoice = event["data"]["object"]
from app import email
email.send(
to=invoice["customer_email"],
subject="Payment failed - action needed",
body="""
<h2>Payment Failed</h2>
<p>We couldn't process your payment. Please update your billing info.</p>
<a href="https://yourapp.com/billing"
style="display:inline-block;background:#f97316;color:#fff;padding:12px 24px;border-radius:6px;text-decoration:none;">
Update Billing
</a>
""",
)
return jsonify({"received": True})import hmac
import hashlib
import time
from flask import Blueprint, request, jsonify, current_app
from emails.receipt import send_receipt
stripe_bp = Blueprint("stripe", __name__)
def verify_stripe_signature(payload: bytes, signature: str, secret: str) -> bool:
parts = dict(item.split("=", 1) for item in signature.split(","))
timestamp = parts.get("t", "")
expected_sig = parts.get("v1", "")
if not timestamp or not expected_sig:
return False
if abs(time.time() - int(timestamp)) > 300:
return False
signed_payload = f"{timestamp}.{payload.decode()}"
computed = hmac.new(
secret.encode(),
signed_payload.encode(),
hashlib.sha256,
).hexdigest()
return hmac.compare_digest(computed, expected_sig)
@stripe_bp.post("/api/webhooks/stripe")
def stripe_webhook():
payload = request.get_data()
signature = request.headers.get("Stripe-Signature", "")
secret = current_app.config["STRIPE_WEBHOOK_SECRET"]
if not verify_stripe_signature(payload, signature, secret):
return jsonify({"error": "Invalid signature"}), 400
event = request.get_json()
if event["type"] == "checkout.session.completed":
session = event["data"]["object"]
from app import email
send_receipt(
email,
to=session["customer_email"],
amount=session["amount_total"],
plan=session.get("metadata", {}).get("plan", "Pro"),
invoice_url=session.get("invoice_url", "https://yourapp.com/billing"),
)
if event["type"] == "invoice.payment_failed":
invoice = event["data"]["object"]
from app import email
email.send(
to=invoice["customer_email"],
subject="Payment failed - action needed",
html="""
<h2>Payment Failed</h2>
<p>We couldn't process your payment. Please update your billing info.</p>
<a href="https://yourapp.com/billing"
style="display:inline-block;background:#f97316;color:#fff;padding:12px 24px;border-radius:6px;text-decoration:none;">
Update Billing
</a>
""",
)
return jsonify({"received": True})import hmac
import hashlib
import time
from flask import Blueprint, request, jsonify, current_app
from emails.receipt import send_receipt
stripe_bp = Blueprint("stripe", __name__)
def verify_stripe_signature(payload: bytes, signature: str, secret: str) -> bool:
parts = dict(item.split("=", 1) for item in signature.split(","))
timestamp = parts.get("t", "")
expected_sig = parts.get("v1", "")
if not timestamp or not expected_sig:
return False
if abs(time.time() - int(timestamp)) > 300:
return False
signed_payload = f"{timestamp}.{payload.decode()}"
computed = hmac.new(
secret.encode(),
signed_payload.encode(),
hashlib.sha256,
).hexdigest()
return hmac.compare_digest(computed, expected_sig)
@stripe_bp.post("/api/webhooks/stripe")
def stripe_webhook():
payload = request.get_data()
signature = request.headers.get("Stripe-Signature", "")
secret = current_app.config["STRIPE_WEBHOOK_SECRET"]
if not verify_stripe_signature(payload, signature, secret):
return jsonify({"error": "Invalid signature"}), 400
event = request.get_json()
if event["type"] == "checkout.session.completed":
session = event["data"]["object"]
from app import email
send_receipt(
email,
to=session["customer_email"],
amount=session["amount_total"],
plan=session.get("metadata", {}).get("plan", "Pro"),
invoice_url=session.get("invoice_url", "https://yourapp.com/billing"),
)
if event["type"] == "invoice.payment_failed":
invoice = event["data"]["object"]
from app import email
email.send(
to=invoice["customer_email"],
subject="Payment failed - action needed",
html="""
<h2>Payment Failed</h2>
<p>We couldn't process your payment. Please update your billing info.</p>
<a href="https://yourapp.com/billing"
style="display:inline-block;background:#f97316;color:#fff;padding:12px 24px;border-radius:6px;text-decoration:none;">
Update Billing
</a>
""",
)
return jsonify({"received": True})Error Handling
import requests
import logging
logger = logging.getLogger(__name__)
class EmailSendError(Exception):
def __init__(self, message: str, status: int, retryable: bool):
super().__init__(message)
self.status = status
self.retryable = retryable
class EmailService:
def __init__(self, api_key: str):
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
})
self.base_url = "https://api.sequenzy.com/v1"
def send(self, to: str, subject: str, body: str) -> dict:
try:
response = self.session.post(
f"{self.base_url}/transactional/send",
json={"to": to, "subject": subject, "body": body},
timeout=30,
)
except requests.Timeout:
raise EmailSendError("Request timed out", 0, retryable=True)
except requests.ConnectionError:
raise EmailSendError("Connection failed", 0, retryable=True)
if response.status_code == 429:
raise EmailSendError("Rate limited", 429, retryable=True)
if response.status_code == 401:
raise EmailSendError("Invalid API key", 401, retryable=False)
if response.status_code >= 500:
raise EmailSendError(f"Server error: {response.text}", response.status_code, retryable=True)
response.raise_for_status()
logger.info(f"Email sent to {to}: {subject}")
return response.json()import requests
import logging
logger = logging.getLogger(__name__)
class EmailSendError(Exception):
def __init__(self, message: str, status: int, retryable: bool):
super().__init__(message)
self.status = status
self.retryable = retryable
class EmailService:
def __init__(self, api_key: str, from_email: str):
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
})
self.base_url = "https://api.resend.com"
self.from_email = from_email
def send(self, to: str, subject: str, html: str) -> dict:
try:
response = self.session.post(
f"{self.base_url}/emails",
json={
"from": self.from_email,
"to": to,
"subject": subject,
"html": html,
},
timeout=30,
)
except requests.Timeout:
raise EmailSendError("Request timed out", 0, retryable=True)
except requests.ConnectionError:
raise EmailSendError("Connection failed", 0, retryable=True)
if response.status_code == 429:
raise EmailSendError("Rate limited", 429, retryable=True)
if response.status_code in (401, 403):
raise EmailSendError("Invalid API key", response.status_code, retryable=False)
if response.status_code >= 500:
raise EmailSendError(f"Server error: {response.text}", response.status_code, retryable=True)
response.raise_for_status()
logger.info(f"Email sent to {to}: {subject}")
return response.json()import requests
import logging
logger = logging.getLogger(__name__)
class EmailSendError(Exception):
def __init__(self, message: str, status: int, retryable: bool):
super().__init__(message)
self.status = status
self.retryable = retryable
class EmailService:
def __init__(self, api_key: str, from_email: str):
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
})
self.base_url = "https://api.sendgrid.com/v3"
self.from_email = from_email
def send(self, to: str, subject: str, html: str) -> dict:
try:
response = self.session.post(
f"{self.base_url}/mail/send",
json={
"personalizations": [{"to": [{"email": to}]}],
"from": {"email": self.from_email},
"subject": subject,
"content": [{"type": "text/html", "value": html}],
},
timeout=30,
)
except requests.Timeout:
raise EmailSendError("Request timed out", 0, retryable=True)
except requests.ConnectionError:
raise EmailSendError("Connection failed", 0, retryable=True)
if response.status_code == 429:
raise EmailSendError("Rate limited", 429, retryable=True)
if response.status_code in (401, 403):
raise EmailSendError("Invalid API key", response.status_code, retryable=False)
if response.status_code >= 500:
raise EmailSendError(f"Server error: {response.text}", response.status_code, retryable=True)
response.raise_for_status()
logger.info(f"Email sent to {to}: {subject}")
return {"sent": True}For critical emails, add a retry wrapper:
# retry.py
import time
from email_service import EmailSendError
def with_retry(fn, max_retries: int = 3):
for attempt in range(max_retries + 1):
try:
return fn()
except EmailSendError as e:
if not e.retryable or attempt == max_retries:
raise
delay = 2 ** attempt
time.sleep(delay)
# Usage
with_retry(lambda: email.send(to=to, subject=subject, body=html))Going to Production
Before you start sending real emails, handle these things.
1. Verify Your Domain
Every email provider requires domain verification. This means adding DNS records (SPF, DKIM, and usually DMARC) to prove you own the domain you're sending from. Our SPF, DKIM, and DMARC setup guide walks through the full process.
Without this, your emails go straight to spam. No exceptions.
2. Use a Dedicated Sending Domain
Send from something like mail.yourapp.com instead of your root domain. If your email reputation takes a hit, your main domain stays clean.
3. Use Gunicorn
Never run flask run in production. Use Gunicorn:
gunicorn app:app -w 4 --bind 0.0.0.0:8000Four workers gives you parallelism. Each worker handles one request at a time (since Flask is sync), so four workers means four concurrent requests.
4. Rate Limit Email Endpoints
Use flask-limiter to prevent abuse:
pip install flask-limiterfrom flask_limiter import Limiter
from flask_limiter.util import get_remote_address
limiter = Limiter(app=app, key_func=get_remote_address)
@app.post("/api/send")
@limiter.limit("10/minute")
def send_email_route():
# ...5. Set Up Logging
import logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)Beyond Transactional: Marketing Emails and Sequences
At some point, you'll want more than one-off transactional emails. You'll want to:
- Send onboarding sequences that guide new users through your product over several days
- Run marketing campaigns to announce features or share updates
- Automate lifecycle emails based on what users do (or don't do) in your app
- Track engagement to see which emails get opened and clicked
Most teams stitch together a transactional provider (Resend, SendGrid) with a separate marketing tool (Mailchimp, ConvertKit). That means two dashboards, two billing systems, and keeping subscriber lists in sync.
Sequenzy handles both from one platform. Same API, same dashboard. You get transactional sends, marketing campaigns, automated sequences, subscriber segments, and native Stripe integration for SaaS-specific automations like trial conversion and churn prevention.
Here's what subscriber management looks like:
import requests
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
# Add a subscriber when they sign up
requests.post("https://api.sequenzy.com/v1/subscribers", headers=headers, json={
"email": "user@example.com",
"firstName": "Jane",
"tags": ["signed-up"],
"customAttributes": {"plan": "free", "source": "organic"},
})
# Tag them when they upgrade
requests.post("https://api.sequenzy.com/v1/subscribers/tags", headers=headers, json={
"email": "user@example.com",
"tag": "customer",
})
# Track events to trigger automated sequences
requests.post("https://api.sequenzy.com/v1/subscribers/events", headers=headers, json={
"email": "user@example.com",
"event": "onboarding.completed",
"properties": {"completedSteps": 5},
})FAQ
Should I use Flask-Mail or an email API?
Use an email API. Flask-Mail uses SMTP, which means you're responsible for deliverability, bounce handling, retry logic, and keeping your mail server healthy. Email APIs handle all of this for you. The only reason to use Flask-Mail is if you have an internal SMTP server you're required to use.
Flask is synchronous. Won't email sends block my app?
Yes, but it's usually fine. An API call to an email provider takes 100-300ms. For most apps, that's acceptable. If it bothers you, use send_in_background (shown above) to offload to a thread, or use Celery for a proper task queue.
When should I use threads vs Celery for background emails?
Use threads for simple apps with low email volume. Threads require no additional infrastructure. Use Celery when you need: retries with backoff, persistent task queues that survive restarts, scheduled tasks, or when your email volume is high enough that you want dedicated worker processes.
Can I use httpx instead of requests for async sending?
Flask itself is synchronous, so using httpx in async mode doesn't help unless you're running Flask with an async server like hypercorn. If you want true async, consider FastAPI instead. For Flask, requests with a Session for connection pooling is the right choice.
How do I test email sending?
Mock the EmailService in tests. Since the service is passed to your email functions, you can swap it with a mock:
from unittest.mock import MagicMock
mock_email = MagicMock()
send_password_reset(mock_email, "user@example.com", "token123")
mock_email.send.assert_called_once()How do I preview email templates during development?
Add a preview route (only in development):
@app.get("/preview/<template_name>")
def preview_email(template_name):
if app.debug:
return render_template(f"emails/{template_name}.html", name="Alice", login_url="#")
return "Not found", 404What about Flask-Mailman?
Flask-Mailman is the Django-style alternative to Flask-Mail. Same problem: it's SMTP-based. For production transactional emails, HTTP APIs are simpler and more reliable.
How do I handle email bounces?
Set up a webhook endpoint to receive delivery notifications from your email provider. Each provider sends events for bounces, complaints, and deliveries. Parse the webhook, update your subscriber status, and stop sending to bounced addresses.
Can I send bulk emails from Flask?
Don't send bulk emails synchronously in Flask routes. Use Celery to queue individual sends, or use your provider's batch API. Flask's synchronous nature means a route that sends 1000 emails would block for minutes.
What's the best way to handle email templates with variables?
Jinja2 template inheritance (shown above) is the Flask way. Define a base template with common styles, extend it for each email type, and pass variables with render_template. Jinja2's auto-escaping prevents XSS in email content too.
Wrapping Up
Here's what we covered:
- Email service class with
requests.Sessionfor connection pooling - Jinja2 templates with inheritance for maintainable email HTML
- Form submissions with Post/Redirect/Get and flash messages
- Background threads for non-blocking sends without infrastructure
- Celery for production task queues with retries
- Blueprints for organizing email routes
- Error handling with custom exceptions and retries
- Stripe webhooks with
hmacsignature verification - Production checklist: Gunicorn, rate limiting, domain verification
The code in this guide is production-ready. Copy the patterns that fit your app, swap in your provider of choice, and start sending.
Frequently Asked Questions
Should I use Flask-Mail or a dedicated email API?
Use a dedicated email API for production. Flask-Mail uses SMTP which requires you to manage connections and delivery yourself. Email APIs provide better deliverability, retry logic, and analytics. Flask-Mail is fine for prototyping or internal tools.
How do I send emails asynchronously in Flask?
Use Celery with Redis as the broker. Create a Celery task for email sending and call it with .delay() from your route handler. For simpler setups, use threading.Thread to send in the background, but this doesn't survive server restarts.
How do I handle email sending in Flask Blueprints?
Import your email function in the Blueprint's route handlers. Keep email logic in a separate services/email.py module that can be imported from any Blueprint. Don't tie email sending to a specific Blueprint—it's shared functionality.
How do I use Jinja2 templates for email content in Flask?
Flask already uses Jinja2 for rendering. Create email templates in templates/emails/ and render them with render_template('emails/welcome.html', name=name). You get full Jinja2 syntax including template inheritance and macros.
How do I test email sending in Flask?
Use Flask's test client and mock the email SDK with unittest.mock.patch. Alternatively, create a fake email backend that captures messages in a list. Assert on the captured messages' recipients, subjects, and content in your tests.
How do I store email API keys in Flask?
Use environment variables loaded with python-dotenv. Access them in your Flask config with os.environ.get('API_KEY') or through Flask's config system. Never hardcode keys in your application files. Use .flaskenv for non-sensitive config and .env for secrets.
Can I send emails from Flask CLI commands?
Yes. Flask CLI commands have full access to the application context. Create custom commands with @app.cli.command() and call your email functions directly. This is useful for batch emails, testing, and administrative notifications.
How do I add rate limiting to Flask email endpoints?
Use Flask-Limiter to decorate your email-sending routes with rate limits like @limiter.limit("5 per minute"). Configure it with Redis as the storage backend for consistency across multiple workers. Rate limit by IP or authenticated user.
How do I handle form submissions that trigger emails in Flask?
Use Flask-WTF for form validation with CSRF protection. In the route handler, validate the form, send the email if valid, and redirect with a flash message. Use the POST-redirect-GET pattern to prevent duplicate submissions on page refresh.
Should I use Flask's send_from_directory for email attachments?
No, that's for serving static files via HTTP. For email attachments, read the file with open() in binary mode, encode it, and pass it to your email SDK's attachment parameter. Keep attachment handling in your email service, not in route handlers.