Sending Email in Python (2026): urllib, httpx, requests, Django, and Webhook Verification

Python ships everything you need for transactional email in its standard library: urllib or http.client for HTTP, hmac and hashlib for HMAC-SHA256, and hmac.compare_digest for timing-safe comparison. In practice, most Python projects also reach for httpx or requests. This guide covers all three, plus Django integration, idempotent retries, file attachments, concurrent batch sends with asyncio, and webhook signature verification.

Prerequisites

• Python 3.10 or newer
• A Rovela API key from the dashboard
• httpx for the async sections (install with pip install httpx)
• Django 4.2 or newer for the Django email backend section

Standard Library: Sending a Single Email

urllib.request is enough for a simple send. Keep the API key in an environment variable.

import json
import os
import urllib.error
import urllib.request
from dataclasses import dataclass, field

ROVELA_URL = "https://api.rovela.dev/emails"


@dataclass
class RovelaError(Exception):
    status_code: int
    name: str
    message: str

    def __str__(self) -> str:
        return f"[{self.status_code}] {self.name}: {self.message}"


def send_email(
    api_key: str,
    payload: dict,
    idempotency_key: str = "",
) -> str:
    body = json.dumps(payload).encode()
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }
    if idempotency_key:
        headers["Idempotency-Key"] = idempotency_key

    req = urllib.request.Request(ROVELA_URL, data=body, headers=headers, method="POST")
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            data = json.loads(resp.read())
            return data["id"]
    except urllib.error.HTTPError as exc:
        data = json.loads(exc.read())
        raise RovelaError(exc.code, data.get("name", "Unknown"), data.get("message", ""))

Usage:

api_key = os.environ["ROVELA_API_KEY"]

email_id = send_email(api_key, {
    "from": "hello@yourcompany.com",
    "to": ["customer@example.com"],
    "subject": "Welcome aboard",
    "html": "<h1>You are in.</h1><p>Click below to get started.</p>",
})
print(f"sent: {email_id}")

Retries with Idempotency-Key

The Rovela API accepts an Idempotency-Key header on POST /emails. Within a 24-hour window, repeating the same key returns the cached response instead of sending a duplicate. Generate the key once before the loop so every retry reuses it. Retry only on 429 Too Many Requests.

import time
import uuid


def send_with_retry(api_key: str, payload: dict, max_attempts: int = 3) -> str:
    # Generated once; reused on every retry so the send is idempotent.
    idempotency_key = str(uuid.uuid4())

    for attempt in range(max_attempts):
        try:
            return send_email(api_key, payload, idempotency_key)
        except RovelaError as exc:
            if exc.status_code != 429:
                raise
            time.sleep(2 ** attempt)

    raise RuntimeError(f"exceeded {max_attempts} attempts")

File Attachments

Add an attachments list to the payload. Each item needs a filename, base64-encoded content, and a content_type.

import base64
from pathlib import Path


def build_invoice_email(recipient: str, invoice_path: Path) -> dict:
    content = base64.b64encode(invoice_path.read_bytes()).decode()
    return {
        "from": "billing@yourcompany.com",
        "to": [recipient],
        "subject": "Your invoice",
        "html": "<p>Please find your invoice attached.</p>",
        "attachments": [
            {
                "filename": "invoice.pdf",
                "content": content,
                "content_type": "application/pdf",
            }
        ],
    }

httpx: Async Concurrent Batch Sends

httpx.AsyncClient pairs with asyncio.gather for concurrent sends. Use asyncio.Semaphore to cap concurrency so you stay under the Rovela rate limit. Generate each idempotency key before gathering so retries inside send_with_retry_async stay stable.

import asyncio
import httpx


async def send_one(
    client: httpx.AsyncClient,
    api_key: str,
    payload: dict,
    idempotency_key: str,
    max_attempts: int = 3,
) -> str:
    for attempt in range(max_attempts):
        resp = await client.post(
            ROVELA_URL,
            json=payload,
            headers={
                "Authorization": f"Bearer {api_key}",
                "Idempotency-Key": idempotency_key,
            },
            timeout=10,
        )
        if resp.status_code == 429:
            await asyncio.sleep(2 ** attempt)
            continue
        resp.raise_for_status()
        return resp.json()["id"]
    raise RuntimeError(f"exceeded {max_attempts} attempts")


async def send_batch(api_key: str, payloads: list[dict], concurrency: int = 5) -> list[str]:
    sem = asyncio.Semaphore(concurrency)
    # Generate idempotency keys before launching tasks so each retry reuses its key.
    keys = [str(uuid.uuid4()) for _ in payloads]

    async def guarded(payload: dict, key: str) -> str:
        async with sem:
            async with httpx.AsyncClient() as client:
                return await send_one(client, api_key, payload, key)

    return await asyncio.gather(*(guarded(p, k) for p, k in zip(payloads, keys)))

Call it from a synchronous context with asyncio.run:

payloads = [
    {"from": "news@yourcompany.com", "to": [addr], "subject": "Weekly digest", "html": html}
    for addr, html in recipients
]
ids = asyncio.run(send_batch(api_key, payloads))

The batch endpoint (POST /emails/batch) accepts up to 100 recipients in a single call and is more efficient when every recipient shares the same subject and body. Reserve the concurrent approach above for personalised sends where each payload differs.

requests: Synchronous Client

For projects already using requests:

import requests


def send_email_requests(api_key: str, payload: dict, idempotency_key: str = "") -> str:
    headers = {"Authorization": f"Bearer {api_key}"}
    if idempotency_key:
        headers["Idempotency-Key"] = idempotency_key

    resp = requests.post(ROVELA_URL, json=payload, headers=headers, timeout=10)
    if resp.status_code >= 400:
        data = resp.json()
        raise RovelaError(resp.status_code, data.get("name", ""), data.get("message", ""))
    return resp.json()["id"]

Django: Custom Email Backend

Django's email layer lets you swap the backend without changing call sites. A custom backend translates Django's EmailMessage objects to Rovela API calls, so existing send_mail and EmailMultiAlternatives usage keeps working.

# myapp/backends.py
import base64
import json
import urllib.error
import urllib.request
from django.core.mail.backends.base import BaseEmailBackend
from django.core.mail import EmailMessage


ROVELA_URL = "https://api.rovela.dev/emails"


class RovelaEmailBackend(BaseEmailBackend):
    def __init__(self, api_key: str | None = None, **kwargs):
        super().__init__(**kwargs)
        from django.conf import settings
        self.api_key = api_key or settings.ROVELA_API_KEY

    def send_messages(self, email_messages: list[EmailMessage]) -> int:
        sent = 0
        for msg in email_messages:
            try:
                self._send_one(msg)
                sent += 1
            except Exception:
                if not self.fail_silently:
                    raise
        return sent

    def _send_one(self, msg: EmailMessage) -> None:
        payload: dict = {
            "from": msg.from_email,
            "to": list(msg.to),
            "subject": msg.subject,
        }
        if msg.cc:
            payload["cc"] = list(msg.cc)
        if msg.bcc:
            payload["bcc"] = list(msg.bcc)
        if msg.reply_to:
            payload["reply_to"] = msg.reply_to[0]

        # Prefer HTML alternative when available.
        html_body = None
        for content, mimetype in getattr(msg, "alternatives", []):
            if mimetype == "text/html":
                html_body = content
                break

        if html_body:
            payload["html"] = html_body
        if msg.body:
            payload["text"] = msg.body

        attachments = []
        for filename, content, mimetype in msg.attachments:
            if isinstance(content, str):
                content = content.encode()
            attachments.append({
                "filename": filename,
                "content": base64.b64encode(content).decode(),
                "content_type": mimetype or "application/octet-stream",
            })
        if attachments:
            payload["attachments"] = attachments

        body = json.dumps(payload).encode()
        req = urllib.request.Request(
            ROVELA_URL,
            data=body,
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json",
            },
            method="POST",
        )
        try:
            with urllib.request.urlopen(req, timeout=10):
                pass
        except urllib.error.HTTPError as exc:
            data = json.loads(exc.read())
            raise RovelaError(exc.code, data.get("name", ""), data.get("message", ""))

Configure it in settings.py:

EMAIL_BACKEND = "myapp.backends.RovelaEmailBackend"
ROVELA_API_KEY = env("ROVELA_API_KEY")

Existing call sites need no changes:

from django.core.mail import send_mail, EmailMultiAlternatives

# Plain text
send_mail("Subject", "Body text.", "from@example.com", ["to@example.com"])

# HTML with text fallback
msg = EmailMultiAlternatives("Subject", "Text fallback.", "from@example.com", ["to@example.com"])
msg.attach_alternative("<p>HTML body</p>", "text/html")
msg.send()

Webhook Signature Verification

Rovela signs webhook deliveries with HMAC-SHA256. The signature is in the x-resend-signature header as t=<timestamp>,v1=<hex-signature>. The signed string is the timestamp, a literal dot, then the raw request body: <timestamp>.<raw-body>.

Python's hmac.new computes the HMAC and hmac.compare_digest does the timing-safe comparison.

import hashlib
import hmac


def verify_webhook(raw_body: bytes, signature_header: str, secret: str) -> dict:
    if not signature_header or "," not in signature_header:
        raise ValueError("missing or malformed x-resend-signature")

    parts: dict[str, str] = {}
    for segment in signature_header.split(","):
        k, _, v = segment.partition("=")
        parts[k] = v

    timestamp = parts.get("t", "")
    v1 = parts.get("v1", "")
    if not timestamp or not v1:
        raise ValueError("malformed x-resend-signature")

    # signed string = "<timestamp>.<raw-body>"
    signed = f"{timestamp}.".encode() + raw_body
    expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest()

    if not hmac.compare_digest(expected, v1):
        raise ValueError("signature mismatch")

    return json.loads(raw_body)

Django Webhook View

Read the raw body before Django's middleware processes it. request.body gives the exact bytes from the request stream.

# myapp/views.py
import json
from django.http import HttpRequest, HttpResponse, HttpResponseBadRequest
from django.views.decorators.csrf import csrf_exempt


@csrf_exempt
def rovela_webhook(request: HttpRequest) -> HttpResponse:
    if request.method != "POST":
        return HttpResponse(status=405)

    secret = settings.ROVELA_WEBHOOK_SECRET
    header = request.headers.get("x-resend-signature", "")

    try:
        event = verify_webhook(request.body, header, secret)
    except ValueError:
        return HttpResponse("Unauthorized", status=401)

    # Dispatch to your task queue.
    # handle_rovela_event.delay(event)

    return HttpResponse(status=204)

Register the route in urls.py:

from django.urls import path
from myapp.views import rovela_webhook

urlpatterns = [
    path("webhooks/rovela/", rovela_webhook),
]

Flask Webhook View

from flask import Flask, request, abort
import os

app = Flask(__name__)

@app.post("/webhooks/rovela")
def rovela_webhook():
    secret = os.environ["ROVELA_WEBHOOK_SECRET"]
    header = request.headers.get("x-resend-signature", "")
    try:
        event = verify_webhook(request.get_data(), header, secret)
    except ValueError:
        abort(401)
    # handle event
    return "", 204

request.get_data() returns the raw bytes before Flask's JSON parsing, which is what you need.

Production Checklist

• API key in environment: read from ROVELA_API_KEY; never hard-code or commit it.
• Idempotency key before the loop: call uuid.uuid4() once before the retry loop, not inside it.
• Retry only on 429: the API returns {"statusCode":429,"name":"Too Many Requests"} when you exceed the send limit; retrying is safe because the idempotency key prevents duplicates.
• Raw body for webhooks: use request.body (Django) or request.get_data() (Flask) before any framework transformation; the signature covers the exact bytes Rovela sent.
• Timing-safe comparison: always use hmac.compare_digest, never == on the hex string.
• Semaphore under the rate limit: cap asyncio.Semaphore concurrency below your Rovela send limit divided by expected per-send latency.
• Webhook secret in environment: store in ROVELA_WEBHOOK_SECRET and rotate it in the dashboard and your deployment atomically.