E-Mails mit Python senden

Diese Anleitung zeigt, wie Sie E-Mails aus Python mit dem offiziellen Paket emailit senden, wie Sie stattdessen SMTP aus Django oder der Standardbibliothek verwenden und wie Sie Webhook-Signaturen in Flask und Django verifizieren.

Voraussetzungen

• Python 3.7 oder neuer.
• Eine verifizierte Versanddomain, zum Beispiel acme.com.
• Ein API-Schlüssel. Ein reiner Sende-Schlüssel, der auf Ihre Domain beschränkt ist, genügt.
• Solange Ihr Workspace keinen Produktionszugang hat, können Sie nur an die Konto-E-Mail-Adressen von Workspace-Mitgliedern senden.

SDK installieren

pip install emailit

Das SDK verwendet intern requests.

API-Schlüssel konfigurieren

Setzen Sie den Schlüssel in der Umgebung Ihres App-Servers, Workers oder Containers:

export EMAILIT_API_KEY=secret_••••••••••••••••••••••••••••••••

Lesen Sie ihn mit os.environ["EMAILIT_API_KEY"], damit ein fehlender Schlüssel schon beim Start deutlich fehlschlägt und nicht erst beim ersten Versand.

E-Mail senden

import os

from emailit import EmailitClient

client = EmailitClient(os.environ["EMAILIT_API_KEY"])

email = client.emails.send({
    "from": "Acme <hello@acme.com>",
    "to": "ada@example.com",
    "subject": "Reset your password",
    "html": "<p>Use this link to reset your password.</p>",
    "text": "Use this link to reset your password.",
})

print(email.id)  # em_…

Erstellen Sie den Client einmal und verwenden Sie ihn wieder. Derselbe Aufruf von send akzeptiert cc, bcc, reply_to, attachments, template mit variables, scheduled_at und tracking; siehe E-Mail senden.

Fehler behandeln

from emailit import (
    ApiErrorException,
    AuthenticationException,
    RateLimitException,
    UnprocessableEntityException,
)

try:
    client.emails.send(message)
except RateLimitException:
    # 429: wait and retry, for example from a task queue
    raise
except AuthenticationException:
    # 401: the API key is missing or invalid
    raise
except UnprocessableEntityException as e:
    # 422: for example, the from domain isn't verified
    print(e.json_body)
except ApiErrorException as e:
    print(e.http_status, e.json_body)

Das SDK ist synchron. Rufen Sie es in asynchronen Frameworks wie FastAPI aus einem normalen def-Endpunkt oder einer Hintergrundaufgabe auf, damit es die Event-Loop nicht blockiert.

Alternativ per SMTP senden

Django

Das integrierte SMTP-Backend von Django funktioniert mit Emailit. Fügen Sie Folgendes zu settings.py hinzu:

import os

EMAIL_BACKEND = "django.core.mail.backends.smtp.EmailBackend"
EMAIL_HOST = "smtp.emailit.com"
EMAIL_PORT = 587
EMAIL_USE_TLS = True
EMAIL_HOST_USER = "emailit"
EMAIL_HOST_PASSWORD = os.environ["EMAILIT_API_KEY"]
DEFAULT_FROM_EMAIL = "Acme <hello@acme.com>"
SERVER_EMAIL = "alerts@acme.com"

Verwenden Sie dann wie gewohnt die Mail-Funktionen von Django:

from django.core.mail import send_mail

send_mail(
    subject="Reset your password",
    message="Use this link to reset your password.",
    from_email=None,  # uses DEFAULT_FROM_EMAIL
    recipient_list=["ada@example.com"],
    html_message="<p>Use this link to reset your password.</p>",
)

DEFAULT_FROM_EMAIL und SERVER_EMAIL müssen eine verifizierte Versanddomain verwenden. Für implizites TLS auf Port 465 setzen Sie EMAIL_PORT = 465 und EMAIL_USE_SSL = True statt EMAIL_USE_TLS.

Standardbibliothek

Ohne Framework verwenden Sie smtplib:

import os
import smtplib
from email.message import EmailMessage

message = EmailMessage()
message["From"] = "Acme <hello@acme.com>"
message["To"] = "ada@example.com"
message["Subject"] = "Reset your password"
message.set_content("Use this link to reset your password.")
message.add_alternative("<p>Use this link to reset your password.</p>", subtype="html")

with smtplib.SMTP("smtp.emailit.com", 587) as smtp:
    smtp.starttls()
    smtp.login("emailit", os.environ["EMAILIT_API_KEY"])
    smtp.send_message(message)

Wenn Ihr Netzwerk 587 blockiert, verwenden Sie Port 2525 oder 2587 mit demselben Code. Siehe SMTP-Einstellungen.

Webhooks empfangen

Erstellen Sie einen Webhook und speichern Sie sein Signatur-Secret in EMAILIT_WEBHOOK_SECRET. Verifizieren Sie die Signatur mit dieser Hilfsfunktion anhand des unveränderten Anfrage-Bodys:

import hashlib
import hmac
import os
import time


def is_valid_signature(raw_body: bytes, signature: str, timestamp: str) -> bool:
    if not signature or not timestamp or not timestamp.isdigit():
        return False
    if abs(time.time() - int(timestamp)) > 300:
        return False
    secret = os.environ["EMAILIT_WEBHOOK_SECRET"].encode()
    expected = hmac.new(secret, timestamp.encode() + b"." + raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)

Rufen Sie sie dann aus der Route Ihres Frameworks auf:

import json

from flask import Flask, abort, request

from emailit_webhooks import is_valid_signature

app = Flask(__name__)


@app.post("/webhooks/emailit")
def emailit_webhook():
    raw_body = request.get_data()
    if not is_valid_signature(
        raw_body,
        request.headers.get("X-Emailit-Signature", ""),
        request.headers.get("X-Emailit-Timestamp", ""),
    ):
        abort(401)

    for event in json.loads(raw_body):  # an array of up to 100 events
        if event["type"] == "email.bounced":
            address = event["data"]["object"]["to"]
            # Stop emailing this address.

    return "", 200

import json

from django.http import HttpResponse, HttpResponseForbidden
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST

from emailit_webhooks import is_valid_signature


@csrf_exempt
@require_POST
def emailit_webhook(request):
    if not is_valid_signature(
        request.body,
        request.headers.get("X-Emailit-Signature", ""),
        request.headers.get("X-Emailit-Timestamp", ""),
    ):
        return HttpResponseForbidden("Invalid signature")

    for event in json.loads(request.body):  # an array of up to 100 events
        if event["type"] == "email.complained":
            address = event["data"]["object"]["to"]
            # Stop emailing this address.

    return HttpResponse(status=200)

Geben Sie innerhalb von 30 Sekunden einen 2xx-Status zurück; bei anderen Antworten wird die Anfrage wiederholt. Siehe Anfragesignatur.

Tipps für den Produktivbetrieb

• Aus einer Task-Queue senden. Verwenden Sie für Massen-E-Mails Celery, RQ oder die Hintergrundaufgaben Ihres Frameworks und drosseln Sie die Worker, um unter Ihren Versandlimits zu bleiben (standardmäßig 2 E-Mails pro Sekunde).
• Wiederholungen sicher machen. Wenn eine Aufgabe nach einem Timeout wiederholt wird, senden Sie einen Header Idempotency-Key (mit requests oder httpx), damit die E-Mail nicht doppelt gesendet wird. Siehe Idempotenz.
• Webhooks deduplizieren. Speichern Sie jede verarbeitete event_id; fehlgeschlagene Zustellungen werden wiederholt und können mehrfach ankommen.
• Einen Schlüssel pro Umgebung verwenden und ihn aus der Umgebung oder einem Secrets-Manager laden. Schreiben Sie ihn nie fest in settings.py.

Nächste Schritte