Tutorial
Enviar emails con Python
Envía emails desde Python con el SDK emailit, el backend SMTP de Django o smtplib, y verifica los webhooks de Emailit en Flask o Django.
En esta guía se explica cómo enviar emails desde Python con el paquete oficial emailit, cómo usar en su lugar SMTP desde Django o desde la biblioteca estándar y cómo verificar las firmas de los webhooks en Flask y Django.
Requisitos previos
- Python 3.7 o posterior.
- Un dominio de envío verificado, por ejemplo
acme.com. - Una clave de API. Basta con una clave Sending Only limitada a tu dominio.
- Hasta que tu espacio de trabajo tenga acceso de producción, solo puedes enviar a las direcciones de email de las cuentas de los miembros del espacio de trabajo.
Instalar el SDK
pip install emailitInternamente, el SDK usa requests.
Configurar la clave de API
Define la clave en el entorno de tu servidor de aplicaciones, de tu worker o de tu contenedor:
export EMAILIT_API_KEY=secret_••••••••••••••••••••••••••••••••Léela con os.environ["EMAILIT_API_KEY"], para que, si falta la clave, se produzca un error claro al arrancar y no en el primer envío.
Enviar 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 el cliente una vez y reutilízalo. La misma llamada a send acepta cc, bcc, reply_to, attachments, template con variables, scheduled_at y tracking; consulta Enviar un email.
Gestionar los errores
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)El SDK es síncrono. En frameworks asíncronos como FastAPI, llámalo desde un endpoint def normal o desde una tarea en segundo plano para que no bloquee el bucle de eventos.
Enviar por SMTP
Django
El backend SMTP integrado de Django funciona con Emailit. Añade esto 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"Después, usa las funciones de correo de Django como de costumbre:
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 y SERVER_EMAIL deben usar un dominio de envío verificado. Para TLS implícito en el puerto 465, define EMAIL_PORT = 465 y EMAIL_USE_SSL = True en lugar de EMAIL_USE_TLS.
Biblioteca estándar
Sin 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)Si tu red bloquea el 587, usa el puerto 2525 o el 2587 con el mismo código. Consulta Configuración SMTP.
Recibir webhooks
Crea un webhook y guarda su secreto de firma en EMAILIT_WEBHOOK_SECRET. Verifica la firma con el cuerpo en bruto de la petición mediante esta función auxiliar:
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)Después, llámala desde la ruta de tu 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)Devuelve un 2xx en menos de 30 segundos; las demás respuestas se reintentan. Consulta Firma de las peticiones.
Consejos para producción
- Envía desde una cola de tareas. Usa Celery, RQ o las tareas en segundo plano de tu framework para los envíos masivos, y limita el ritmo de los workers para no superar tus límites de envío (2 emails por segundo por defecto).
- Haz que los reintentos sean seguros. Cuando una tarea se reintente después de que se agote el tiempo de espera, envía una cabecera
Idempotency-Key(conrequestsohttpx) para que el email no se envíe dos veces. Consulta Idempotencia. - Elimina los webhooks duplicados. Guarda cada
event_idque procesas; las entregas fallidas se reintentan y pueden llegar más de una vez. - Usa una clave por entorno y cárgala desde el entorno o desde un gestor de secretos. No la escribas nunca directamente en
settings.py.