Tutorial
Invia email con Python
Invia email da Python con l’SDK emailit, con il backend SMTP di Django o con smtplib, e verifica i webhook di Emailit in Flask o Django.
Questa guida mostra come inviare email da Python con il pacchetto ufficiale emailit, come usare in alternativa SMTP da Django o dalla libreria standard e come verificare le firme dei webhook in Flask e Django.
Prerequisiti
- Python 3.7 o versioni successive.
- Un dominio di invio verificato, ad esempio
acme.com. - Una chiave API. Basta una chiave di solo invio limitata al tuo dominio.
- Finché il workspace non ha l’accesso alla produzione, puoi inviare solo agli indirizzi email degli account dei membri del workspace.
Installa l’SDK
pip install emailitInternamente l’SDK usa requests.
Configura la chiave API
Imposta la chiave nell’ambiente del server dell’app, del worker o del container:
export EMAILIT_API_KEY=secret_••••••••••••••••••••••••••••••••Leggila con os.environ["EMAILIT_API_KEY"], così una chiave mancante genera subito un errore all’avvio invece che al primo invio.
Invia un’email
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_…Crea il client una sola volta e riutilizzalo. La stessa chiamata send accetta cc, bcc, reply_to, attachments, template con variables, scheduled_at e tracking; vedi Invia un’email.
Gestisci gli errori
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)L’SDK è sincrono. Nei framework asincroni come FastAPI, chiamalo da un normale endpoint def o da un task in background, così non blocca l’event loop.
Invia con SMTP
Django
Il backend SMTP integrato di Django funziona con Emailit. Aggiungi questo a 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"Poi usa le funzioni di posta di Django come al solito:
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 e SERVER_EMAIL devono usare un dominio di invio verificato. Per il TLS implicito sulla porta 465, imposta EMAIL_PORT = 465 e EMAIL_USE_SSL = True al posto di EMAIL_USE_TLS.
Libreria standard
Senza un framework, usa 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)Se la tua rete blocca la 587, usa la porta 2525 o 2587 con lo stesso codice. Vedi Impostazioni SMTP.
Ricevi i webhook
Crea un webhook e salva il suo secret di firma in EMAILIT_WEBHOOK_SECRET. Verifica la firma sul corpo grezzo della richiesta con questa funzione di supporto:
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)Poi chiamala dalla route del tuo framework:
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)Restituisci un 2xx entro 30 secondi; le altre risposte vengono ritentate. Vedi Firma delle richieste.
Consigli per la produzione
- Invia da una coda di task. Per gli invii massivi usa Celery, RQ o i task in background del tuo framework, e limita la velocità dei worker per restare entro i limiti di invio (2 email al secondo per impostazione predefinita).
- Rendi sicuri i nuovi tentativi. Quando un task viene ritentato dopo un timeout, invia un header
Idempotency-Key(conrequestsohttpx), così l’email non viene inviata due volte. Vedi Idempotenza. - Elimina i webhook duplicati. Salva ogni
event_idche elabori; le consegne non riuscite vengono ritentate e possono arrivare più di una volta. - Usa una chiave per ogni ambiente e caricala dall’ambiente o da un gestore di secret. Non scriverla mai direttamente in
settings.py.