How to Send Emails in Flask (2026 Guide)

pip install flask requests python-dotenv

pip install flask requests python-dotenv

pip install flask requests python-dotenv

SEQUENZY_API_KEY=sq_your_api_key_here

RESEND_API_KEY=re_your_api_key_here

SENDGRID_API_KEY=SG.your_api_key_here

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}

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"],
)

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})

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})

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")

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"})

from 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)

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,
  )

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,
  )

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})

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}