Tutoriál
Odesílání e-mailů z Pythonu
Odesílejte e-maily z Pythonu přes SDK emailit, SMTP backend Djanga nebo smtplib a ověřujte webhooky Emailitu ve Flasku nebo Djangu.
Tento návod ukazuje, jak odesílat e-maily z Pythonu přes oficiální balíček emailit, jak místo toho použít SMTP z Djanga nebo ze standardní knihovny a jak ověřovat podpisy webhooků ve Flasku a Djangu.
Předpoklady
- Python 3.7 nebo novější.
- Ověřená odesílací doména, například
acme.com. - API klíč. Stačí klíč jen pro odesílání omezený na vaši doménu.
- Dokud váš workspace nemá produkční přístup, můžete odesílat jen na e-mailové adresy účtů členů workspace.
Nainstalujte SDK
pip install emailitSDK interně používá requests.
Nastavte API klíč
Nastavte klíč v prostředí svého aplikačního serveru, workeru nebo kontejneru:
export EMAILIT_API_KEY=secret_••••••••••••••••••••••••••••••••Načítejte ho pomocí os.environ["EMAILIT_API_KEY"], aby chybějící klíč způsobil viditelnou chybu už při spuštění, a ne až při prvním odeslání.
Odešlete e-mail
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_…Klienta vytvořte jednou a používejte ho opakovaně. Stejné volání send přijímá cc, bcc, reply_to, attachments, template s variables, scheduled_at a tracking; viz Odeslání e-mailu.
Ošetřete chyby
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)SDK je synchronní. V asynchronních frameworcích, jako je FastAPI, ho volejte z běžného endpointu def nebo z úlohy na pozadí, aby neblokovalo smyčku událostí.
Odesílání přes SMTP
Django
Vestavěný SMTP backend Djanga s Emailitem funguje. Přidejte do settings.py:
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"Pak používejte e-mailové funkce Djanga jako obvykle:
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 a SERVER_EMAIL musí používat ověřenou odesílací doménu. Pro implicitní TLS na portu 465 nastavte místo EMAIL_USE_TLS hodnoty EMAIL_PORT = 465 a EMAIL_USE_SSL = True.
Standardní knihovna
Bez frameworku použijte 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)Pokud vaše síť blokuje port 587, použijte se stejným kódem port 2525 nebo 2587. Viz Nastavení SMTP.
Přijímejte webhooky
Vytvořte webhook a jeho tajný klíč uložte do EMAILIT_WEBHOOK_SECRET. Podpis ověřte vůči surovému tělu požadavku touto pomocnou funkcí:
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)Pak ji zavolejte z routy svého frameworku:
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 "", 200import 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)Do 30 sekund vraťte 2xx; při jiných odpovědích se požadavek opakuje. Viz Ověření podpisu webhooků.
Tipy pro produkční provoz
- Odesílejte z fronty úloh. Pro hromadnou poštu použijte Celery, RQ nebo úlohy na pozadí svého frameworku a zpomalte workery, abyste zůstali pod svými limity odesílání (ve výchozím stavu 2 e-maily za sekundu).
- Zajistěte bezpečné opakování. Když se úloha po vypršení časového limitu opakuje, pošlete hlavičku
Idempotency-Key(přesrequestsnebohttpx), aby se e-mail neodeslal dvakrát. Viz Idempotentní požadavky. - Odstraňujte duplicitní webhooky. Ukládejte si každé zpracované
event_id; neúspěšná doručení se opakují a mohou dorazit víckrát. - Pro každé prostředí používejte jeden klíč a načítejte ho z prostředí nebo ze správce tajných údajů. Nikdy ho nezapisujte napevno do
settings.py.