Python 发邮件的两种方式：HTTP API（新项目推荐）和 SMTP。覆盖退避重试、幂等键、附件、Django 集成与错误处理。

用 Python 发送邮件

Python 是发送事务邮件最流行的语言之一——标准库自带 smtplib 和 email，而通过 requests 走 HTTP API 对现代应用来说更简单。本文覆盖两种方式、各自的适用场景，以及生产环境中的实际问题：认证、附件、重试和错误处理。

两种方式

HTTP API——向 Rovela 的 REST 端点发 POST 请求。无需配置邮件服务器，适用于任何环境（Serverless、Docker、CI），还能获得结构化的错误响应、投递 Webhook 和幂等键。新项目推荐这种方式。

SMTP——连接邮件服务器转发邮件。适用于只支持 SMTP 的现有库或框架（如 Django 的邮件后端），或需要通过自建服务器发信的场景。

通过 HTTP API 发送

pip install requests

最简发送：

import requests

resp = requests.post(
    "https://api.rovela.dev/emails",
    headers={"Authorization": "Bearer YOUR_API_KEY"},
    json={
        "from": "no-reply@yourdomain.com",
        "to": ["user@example.com"],
        "subject": "您的订单已发货",
        "html": "<p>您的订单 <strong>#1234</strong> 已发出。</p>",
        "text": "您的订单 #1234 已发出。",
    },
)
resp.raise_for_status()
print(resp.json()["id"])

始终同时提供 html 和 text。很多邮件客户端在预览中显示纯文本版本，部分用户偏好纯文本；缺少 text 会降低垃圾邮件过滤器的评分。

生产环境实践

凭据放环境变量

不要硬编码 API Key：

import os, requests

API_KEY = os.environ["ROVELA_API_KEY"]

def send_email(to, subject, html, text):
    resp = requests.post(
        "https://api.rovela.dev/emails",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={"from": "no-reply@yourdomain.com", "to": to,
              "subject": subject, "html": html, "text": text},
        timeout=10,
    )
    resp.raise_for_status()
    return resp.json()["id"]

指数退避重试

网络错误和瞬时 5xx 会发生，用重试循环包裹发送：

import time, random

def send_with_retry(payload, max_attempts=3):
    for attempt in range(max_attempts):
        try:
            resp = requests.post(
                "https://api.rovela.dev/emails",
                headers={"Authorization": f"Bearer {API_KEY}"},
                json=payload, timeout=10,
            )
            if resp.status_code < 500:
                resp.raise_for_status()
                return resp.json()["id"]
        except requests.RequestException:
            pass
        if attempt < max_attempts - 1:
            time.sleep(2 ** attempt + random.random())
    raise RuntimeError("重试后仍发送失败")

4xx 错误不要重试——422（校验错误）或 401（Key 错误）重试多少次都不会成功。

幂等键

如果应用可能对同一事件调用多次发送（重试的后台任务、重复的 Webhook），传入幂等键：

import hashlib

idem_key = hashlib.sha256(f"order-shipped-{order_id}".encode()).hexdigest()[:32]
requests.post(url, headers={
    "Authorization": f"Bearer {API_KEY}",
    "Idempotency-Key": idem_key,
}, json=payload)

Rovela 会在 24 小时内对相同幂等键的请求去重。

发送附件

import base64, pathlib

pdf_bytes = pathlib.Path("invoice.pdf").read_bytes()

requests.post(url, headers=headers, json={
    "from": "billing@yourdomain.com",
    "to": ["customer@example.com"],
    "subject": "您的发票",
    "html": "<p>发票请见附件。</p>",
    "text": "发票请见附件。",
    "attachments": [{
        "filename": "invoice.pdf",
        "content": base64.b64encode(pdf_bytes).decode(),
        "content_type": "application/pdf",
    }],
})

附件大小要合理——大多数服务商拒绝超过 10 MB 的邮件，大附件还会提高垃圾邮件评分。大文件建议在邮件正文中放签名链接。

通过 SMTP 发送

适用于只支持 SMTP 的框架或通过自建中继发信的场景：

import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText

msg = MIMEMultipart("alternative")
msg["Subject"] = "测试邮件"
msg["From"] = "no-reply@yourdomain.com"
msg["To"] = "recipient@example.com"
msg.attach(MIMEText("纯文本版本。", "plain"))
msg.attach(MIMEText("<p>HTML 版本。</p>", "html"))

with smtplib.SMTP_SSL("smtp.rovela.dev", 465) as s:
    s.login("no-reply@yourdomain.com", "YOUR_SMTP_PASSWORD")
    s.send_message(msg)

端口 465 使用隐式 TLS，端口 587 使用 STARTTLS：

with smtplib.SMTP("smtp.rovela.dev", 587) as s:
    s.ehlo()
    s.starttls()
    s.login("no-reply@yourdomain.com", "YOUR_SMTP_PASSWORD")
    s.send_message(msg)

Django 集成

Django 默认用 SMTP，配置 Rovela 的 SMTP 后端：

# settings.py
EMAIL_BACKEND = "django.core.mail.backends.smtp.EmailBackend"
EMAIL_HOST = "smtp.rovela.dev"
EMAIL_PORT = 465
EMAIL_USE_SSL = True
EMAIL_HOST_USER = "no-reply@yourdomain.com"
EMAIL_HOST_PASSWORD = env("ROVELA_SMTP_PASSWORD")
DEFAULT_FROM_EMAIL = "no-reply@yourdomain.com"

然后在应用任意位置发送：

from django.core.mail import send_mail

send_mail(
    "主题",
    "纯文本正文",
    "no-reply@yourdomain.com",
    ["recipient@example.com"],
    html_message="<p>HTML 正文</p>",
)

超时、连接与并发

三个参数值得在封装函数里定死，而不是每次调用临时决定：

超时。 requests 默认不设超时，意味着对端不响应时你的请求会一直挂着，直到操作系统层面的 TCP 超时——可能是几分钟。在 Web 请求的调用链里，这等于把整个请求线程一起拖死。timeout=10 是常见取值：连接超时和读取超时都覆盖到。

连接复用。 每次 requests.post 都新建一个 Session 是浪费。在长期运行的服务里建一个模块级 session = requests.Session()，让底层连接池复用 TLS 握手。发信量大的时候，这一项能省掉可观的延迟。

并发上限。 批量发送时不要 ThreadPoolExecutor(max_workers=500) 一把梭。上游的限流是按秒计的，打满之后你会收到一堆 429，然后全部进入重试队列，反而更慢。用 max_workers=10 到 20，配合 429 的 Retry-After 退避，整体吞吐更稳定。

from concurrent.futures import ThreadPoolExecutor

def send_batch(recipients, payload_for):
    with ThreadPoolExecutor(max_workers=10) as pool:
        futures = {pool.submit(send_with_retry, payload_for(r)): r for r in recipients}
        results = {}
        for fut in futures:
            addr = futures[fut]
            try:
                results[addr] = fut.result()
            except Exception as exc:
                results[addr] = f"failed: {exc}"
    return results

每一封的失败都被单独捕获，一封发不出去不会影响其余的。这一点在批量发送里比什么都重要——一个未捕获的异常会让整批停在中途，而你已经不知道哪些发出去了。

接收投递事件

发出去不等于送达。要拿到送达、退信、打开、点击这些事件，需要在 Rovela 后台配置 Webhook，然后在你的应用里暴露一个端点接收。

验签必须用原始请求体，不能用解析后的 JSON。框架解析再重新序列化之后，字节顺序、空格、转义都会变，签名一定对不上。Flask 里是这样：

from flask import Flask, request, abort
import hmac, hashlib

app = Flask(__name__)

@app.post("/webhooks/rovela")
def rovela_webhook():
    signature = request.headers.get("X-Rovela-Signature", "")
    raw = request.get_data()          # 原始字节，不能先 request.get_json()
    expected = hmac.new(
        WEBHOOK_SECRET.encode(), raw, hashlib.sha256
    ).hexdigest()
    if not hmac.compare_digest(signature, expected):
        abort(401)
    event = request.get_json()
    handle_event(event)               # 用 event["id"] 去重
    return "", 204

两个容易忽略的点：

hmac.compare_digest 而不是 ==，避免时序攻击。
Webhook 会重投。同一个 event["id"] 可能到达两次，处理逻辑必须幂等，否则用户会收到重复通知。

SMTP 的返回值怎么读

用 SMTP 时，smtplib 抛出的异常类型能直接告诉你该不该重试：

异常	含义	处理
`SMTPAuthenticationError`	认证失败	密码或授权码错误，不重试
`SMTPRecipientsRefused`	收件人被拒	地址无效或对方拒收，不重试
`SMTPSenderRefused`	发件人被拒	域名未验证或发件地址不允许
`SMTPDataError`	内容被拒	内容或头部不合规
`SMTPServerDisconnected`	连接中断	可重试
`OSError` / `socket.timeout`	网络层	可重试

SMTPRecipientsRefused 的异常对象里带着每个地址的 (code, message)。把它记下来，550 开头的通常是地址不存在，可以直接加入抑制名单；450/451 是临时故障，值得下次再试。

什么时候该从 SMTP 迁到 API

已经在用 SMTP 的项目不必着急改。出现下面任意一条时，迁移的收益就超过成本了：

需要按消息查询投递状态，而现在只能靠猜或者看日志。
需要幂等重试，而 SMTP 本身没有幂等语义，重发就是重复投递。
部署在 Serverless 或容器里，出站 25 端口被云厂商封掉。
需要抑制名单、退信分类、投诉处理这些平台能力，而不想自己维护。

迁移路径可以渐进：先把新代码路径切到 API，老路径保持不动；等新路径跑稳了，再把老的逐个替换。两种方式可以同时存在，同一个域名既配 SPF 的 include:mail.rovela.dev，也配 SMTP 授权，互不冲突。

错误处理参考

状态码	含义	处理方式
200/201	已发送	读取响应中的 `id`
400	请求格式错误	修正请求，不要重试
401	API Key 无效	检查环境变量，必要时轮换 Key
422	校验错误	读取错误详情，修正后重新提交
429	限流	退避后按 `Retry-After` 秒数重试
5xx	服务器错误	退避重试

成功时记录 Rovela 消息 ID（resp.json()["id"]）。用户反馈邮件未收到时，可用该 ID 在 Rovela 后台查询投递事件。