Python 发送邮件（2026）：urllib、httpx、requests、Django 集成与 Webhook 验签

Python 标准库自带发送事务邮件所需的一切：urllib 或 http.client 处理 HTTP、hmac 和 hashlib 计算 HMAC-SHA256、hmac.compare_digest 做常量时间比较。实际项目中也常用 httpx 或 requests。本文覆盖这三种方式，以及 Django 集成、幂等重试、文件附件、用 asyncio 并发批量发送，和 Webhook 签名验证。

前置条件

• Python 3.10 及以上
• Rovela 控制台里的 API Key
• httpx（异步部分，pip install httpx 安装）
• Django 4.2 及以上（Django 邮件后端部分）

标准库：发送单封邮件

urllib.request 足够完成简单的发送。把 API Key 放在环境变量里。

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

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

调用示例：

api_key = os.environ["ROVELA_API_KEY"]

email_id = send_email(api_key, {
    "from": "hello@yourcompany.com",
    "to": ["customer@example.com"],
    "subject": "欢迎使用平台",
    "html": "<h1>您已成功注册。</h1><p>点击下方按钮开始使用。</p>",
})
print(f"已发送: {email_id}")

带幂等键的重试

Rovela API 在 POST /emails 上接受 Idempotency-Key 请求头。24 小时窗口内，相同的 key 返回缓存响应而不会重复发送。在循环之前生成一次 key，每次重试复用同一个值。只对 429 Too Many Requests 重试。

import time
import uuid


def send_with_retry(api_key: str, payload: dict, max_attempts: int = 3) -> str:
    # 生成一次，每次重试复用，保证幂等
    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"超过最大重试次数 {max_attempts}")

文件附件

在 payload 里加入 attachments 列表。每个附件需要 filename、base64 编码的 content 和 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": "您的发票",
        "html": "<p>请查收附件中的发票。</p>",
        "attachments": [
            {
                "filename": "invoice.pdf",
                "content": content,
                "content_type": "application/pdf",
            }
        ],
    }

httpx：异步并发批量发送

httpx.AsyncClient 配合 asyncio.gather 实现并发发送。用 asyncio.Semaphore 限制并发数，控制在 Rovela 速率限制以内。在 gather 之前为每个请求生成幂等键，保证 send_one 内部的重试稳定。

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"超过最大重试次数 {max_attempts}")


async def send_batch(api_key: str, payloads: list[dict], concurrency: int = 5) -> list[str]:
    sem = asyncio.Semaphore(concurrency)
    # 在启动任务前生成幂等键，每次重试复用同一个
    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)))

在同步代码里用 asyncio.run 调用：

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

批量接口（POST /emails/batch）支持单次最多 100 个收件人，适合所有收件人共享同一正文的场景。只有 payload 因人而异时，才用上面的并发方式。

requests：同步客户端

对于已经依赖 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：自定义邮件后端

Django 的邮件层支持自定义后端，无需修改现有调用点。自定义后端把 Django 的 EmailMessage 对象翻译成 Rovela API 请求，send_mail 和 EmailMultiAlternatives 的用法保持不变。

# 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]

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

在 settings.py 里配置：

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

现有调用点无需修改：

from django.core.mail import send_mail, EmailMultiAlternatives

# 纯文本
send_mail("主题", "正文文本", "from@example.com", ["to@example.com"])

# HTML 加文本兜底
msg = EmailMultiAlternatives("主题", "文本兜底", "from@example.com", ["to@example.com"])
msg.attach_alternative("<p>HTML 正文</p>", "text/html")
msg.send()

Webhook 签名验证

Rovela 用 HMAC-SHA256 对 Webhook 投递签名。签名在 x-resend-signature 头中，格式为 t=<时间戳>,v1=<十六进制签名>。签名原文是时间戳、一个英文句号、原始请求体拼接而成：<时间戳>.<原始请求体>。

Python 的 hmac.new 计算 HMAC，hmac.compare_digest 做常量时间比较。

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("缺少或格式错误的 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("x-resend-signature 头格式错误")

    # 签名原文 = "<时间戳>.<原始请求体>"
    signed = f"{timestamp}.".encode() + raw_body
    expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest()

    if not hmac.compare_digest(expected, v1):
        raise ValueError("签名不匹配")

    return json.loads(raw_body)

Django 视图

在 Django 中间件处理之前读取原始请求体。request.body 返回请求流的确切字节。

from django.http import HttpRequest, HttpResponse
from django.views.decorators.csrf import csrf_exempt


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

    header = request.headers.get("x-resend-signature", "")
    try:
        event = verify_webhook(request.body, header, settings.ROVELA_WEBHOOK_SECRET)
    except ValueError:
        return HttpResponse("Unauthorized", status=401)

    # 分发到任务队列
    # handle_rovela_event.delay(event)

    return HttpResponse(status=204)

在 urls.py 里注册：

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

Flask 视图

from flask import Flask, request, abort

app = Flask(__name__)

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

request.get_data() 返回 Flask JSON 解析前的原始字节，这正是验签所需的。

上线前检查清单

• API Key 放在环境变量里：从 ROVELA_API_KEY 读取，不要硬编码，不要提交到版本控制。
• 幂等键在循环外生成：调用 uuid.uuid4() 一次在 for 循环之前，循环内复用同一个值。
• 只对 429 重试：API 在超限时返回 {"statusCode":429,"name":"Too Many Requests"}；因为有幂等键防止重复，对 4xx 和 5xx 重试也是安全的。
• Webhook 用原始请求体：Django 用 request.body，Flask 用 request.get_data()，在任何框架变换之前读取。
• 使用常量时间比较：签名字节必须用 hmac.compare_digest，不能用 ==。
• Semaphore 低于速率限制：asyncio.Semaphore 的并发数不要超过 Rovela 发送上限除以预期单封延迟的结果。
• Webhook 密钥放在环境变量里：存在 ROVELA_WEBHOOK_SECRET 中，在控制台和部署里同步更换。