Python

Frameworks

Django, Flask y FastAPI, un endpoint de webhooks y trabajos en segundo plano que nunca envían dos veces.

Django

Guarda la clave en la configuración, leída del entorno, y construye un único OpenEmail en un módulo propio que importen las vistas. El cliente se puede compartir entre hilos sin riesgo, así que una sola instancia sirve todas las solicitudes y mantiene un único pool de conexiones para el proceso. Llama al módulo como quieras salvo openemail.py, que puede ocultar el paquete.

settings.py

import os OPENEMAIL_API_KEY = os.environ['OPENEMAIL_API_KEY']OPENEMAIL_TIMEOUT = 10.0OPENEMAIL_SENDER = 'Acme <[email protected]>'

mailer.py

from django.conf import settingsfrom openemail import OpenEmail mailer = OpenEmail(settings.OPENEMAIL_API_KEY, timeout=settings.OPENEMAIL_TIMEOUT)

views.py

from django.conf import settingsfrom django.http import HttpRequest, JsonResponsefrom django.views.decorators.http import require_POSTfrom openemail import OpenEmailApiError from acme.mailer import mailer  @require_POSTdef invite(request: HttpRequest) -> JsonResponse:    try:        sent = mailer.emails.send({            'from': settings.OPENEMAIL_SENDER,            'to': request.POST['email'],            'subject': 'You are invited to Acme',            'text': 'Accept the invitation to join the workspace.',        })    except OpenEmailApiError as error:        if error.is_validation:            return JsonResponse({'error': error.message, 'field': error.param}, status=422)         raise     return JsonResponse({'id': sent['id']}, status=202)

El cliente openemail incluido también sirve aquí: llama a init(settings.OPENEMAIL_API_KEY) una vez, desde el método ready() de la configuración de tu app, e importa openemail allí donde envíes.

Flask

La fábrica de aplicaciones construye el cliente junto con la app y lo guarda en app.extensions, y las vistas acceden a él a través de current_app. Una fábrica que también acepta un cliente permite que una prueba pase uno construido sobre httpx.MockTransport.

app.py

import os from flask import Flask, current_app, requestfrom openemail import OpenEmail  def create_app(mailer: OpenEmail | None = None) -> Flask:    app = Flask(__name__)    app.extensions['openemail'] = mailer or OpenEmail(os.environ['OPENEMAIL_API_KEY'], timeout=10)     @app.post('/invites')    def invite() -> tuple[dict[str, str], int]:        client: OpenEmail = current_app.extensions['openemail']        sent = client.emails.send({            'from': 'Acme <[email protected]>',            'to': request.form['email'],            'subject': 'You are invited to Acme',            'text': 'Accept the invitation to join the workspace.',        })         return {'id': sent['id']}, 202     return app

FastAPI

Construye un único AsyncOpenEmail en el lifespan, para que pertenezca al bucle de eventos que atiende las solicitudes y se cierre cuando se detenga el servidor, y pásalo a las rutas con una dependencia.

main.py

from collections.abc import AsyncIteratorfrom contextlib import asynccontextmanagerfrom typing import Annotated from fastapi import Body, Depends, FastAPI, Requestfrom openemail import AsyncOpenEmail  @asynccontextmanagerasync def lifespan(app: FastAPI) -> AsyncIterator[None]:    async with AsyncOpenEmail(timeout=10) as mailer:        app.state.openemail = mailer        yield  app = FastAPI(lifespan=lifespan)  def get_openemail(request: Request) -> AsyncOpenEmail:    mailer: AsyncOpenEmail = request.app.state.openemail    return mailer  Mailer = Annotated[AsyncOpenEmail, Depends(get_openemail)]  @app.post('/invites', status_code=202)async def invite(email: Annotated[str, Body(embed=True)], mailer: Mailer) -> dict[str, str]:    sent = await mailer.emails.send({        'from': 'Acme <[email protected]>',        'to': email,        'subject': 'You are invited to Acme',        'text': 'Accept the invitation to join the workspace.',    })     return {'id': sent['id'], 'status': sent['status']}

Una prueba sustituye el cliente mediante app.dependency_overrides[get_openemail], así que ninguna ruta llega a la API.

Endpoints de webhook

Verifica cada entrega antes de actuar sobre ella, con el cuerpo sin procesar y las cabeceras de la solicitud: await request.body() y request.headers en FastAPI, request.body y request.headers en Django, y request.get_data() y request.headers en Flask. La búsqueda de la cabecera no distingue mayúsculas de minúsculas, así que el objeto de cabeceras propio de cada framework funciona tal cual.

webhooks.py

import os from fastapi import FastAPI, HTTPException, Request, Responsefrom openemail import WEBHOOK_EVENTS, WebhookVerificationError, verify_webhook_signature from acme.jobs import handle_bounce app = FastAPI()  @app.post('/webhooks/openemail', status_code=204)async def openemail_webhook(request: Request) -> Response:    try:        event = verify_webhook_signature(            payload=await request.body(),            headers=request.headers,            secret=os.environ['OPENEMAIL_WEBHOOK_SECRET'],        )    except WebhookVerificationError as error:        raise HTTPException(status_code=400, detail='bad signature') from error     if event['type'] == WEBHOOK_EVENTS.EMAIL_BOUNCED:        handle_bounce.delay(event['id'], event['data'])     return Response(status_code=204)

webhook_views.py

import os from django.http import HttpRequest, HttpResponsefrom django.views.decorators.csrf import csrf_exemptfrom django.views.decorators.http import require_POSTfrom openemail import WebhookVerificationError, verify_webhook_signature from acme.models import ReceivedEvent  @csrf_exempt@require_POSTdef openemail_webhook(request: HttpRequest) -> HttpResponse:    try:        event = verify_webhook_signature(            payload=request.body,            headers=request.headers,            secret=os.environ['OPENEMAIL_WEBHOOK_SECRET'],        )    except WebhookVerificationError:        return HttpResponse('bad signature', status=400)     ReceivedEvent.objects.get_or_create(        id=event['id'],        defaults={'type': event['type'], 'data': event['data']},    )     return HttpResponse(status=204)

Django rechaza un POST que no lleva token CSRF, y una entrega no lleva ninguno, así que la vista es csrf_exempt: la firma es lo que demuestra que la solicitud vino de OpenEmail. Lee request.headers en lugar de request.META, cuyas claves se renombran al formato HTTP_X_OPENEMAIL_SIGNATURE.

Responde rápido con un 2xx y haz el trabajo después. Una entrega que no recibe respuesta, o que recibe un 408, 425, 429 o 5xx, se vuelve a intentar, hasta 8 veces en unas 27 horas y media, y un reenvío manda de nuevo un evento con el mismo id, así que guarda los ids que ya has procesado y omite las repeticiones.

Verificar una entregaQué rechaza la comprobación de la firma y por qué.EndpointsRegistra un endpoint, pruébalo y reenvía una entrega.

Trabajos en segundo plano

Una cola de trabajos vuelve a ejecutar una tarea cuando falla, y una tarea puede fallar después de que su correo haya salido: se perdió la respuesta, o el worker se detuvo antes de terminar. Pasa un idempotency_key= derivado de aquello que hizo necesario el envío. Así, cada ejecución de la tarea lleva la misma clave, y una repetición reproduce el mensaje original en lugar de enviar uno segundo.

tasks.py

from celery import Task, shared_taskfrom openemail import OpenEmail, OpenEmailApiError, OpenEmailNetworkError mailer = OpenEmail()  @shared_task(bind=True, acks_late=True, max_retries=5)def send_receipt(self: Task, order_id: str, email: str) -> str:    try:        sent = mailer.emails.send(            {                'from': 'Acme Billing <[email protected]>',                'to': email,                'subject': f'Receipt for order {order_id}',                'text': f'Thank you for order {order_id}.',            },            idempotency_key=f'receipt:{order_id}',        )    except OpenEmailNetworkError as error:        raise self.retry(exc=error, countdown=30)    except OpenEmailApiError as error:        if error.is_server_error:            raise self.retry(exc=error, countdown=30)         raise     return sent['id']

Con acks_late=True, Celery confirma una tarea solo cuando se ha ejecutado, así que una tarea interrumpida por un worker que se detuvo puede entregarse de nuevo, lo que aquí no supone ningún riesgo porque la clave convierte esa segunda ejecución en una reproducción. El Retry de RQ vuelve a ejecutar un trabajo fallido, con el mismo efecto.

jobs.py

from redis import Redisfrom rq import Queue, Retry from openemail import OpenEmail mailer = OpenEmail()  def send_receipt(order_id: str, email: str) -> str:    sent = mailer.emails.send(        {            'from': 'Acme Billing <[email protected]>',            'to': email,            'subject': f'Receipt for order {order_id}',            'text': f'Thank you for order {order_id}.',        },        idempotency_key=f'receipt:{order_id}',    )     return sent['id']  queue = Queue(connection=Redis())queue.enqueue(send_receipt, 'AC-4192', '[email protected]', retry=Retry(max=5, interval=[10, 60, 300]))

Deduce la clave de aquello que hizo necesario el envío, nunca de un reloj. Tiene de 1 a 255 letras, dígitos, guiones bajos, puntos, dos puntos o guiones, así que constrúyela a partir de un id y no de una dirección de correo. Construye también el cuerpo solo a partir de los argumentos de la tarea: una repetición con la misma clave y un cuerpo distinto se rechaza con 422 idempotency_key_reuse en lugar de reproducirse.

Construye el cliente a nivel de módulo. No abre ninguna conexión hasta su primera solicitud, así que cada proceso worker creado con fork a partir del padre abre la suya.

Reintentos e idempotenciaQué reintenta el cliente y por qué un envío reintentado no puede salir dos veces.