Python

Frameworks

Django, Flask et FastAPI, un endpoint de webhook, et des tâches en arrière-plan qui n'envoient jamais deux fois.

Django

Gardez la clé dans les réglages, lue depuis l'environnement, et construisez un seul OpenEmail dans un module dédié que les vues importent. Le client peut être partagé sans risque entre threads : une seule instance sert donc chaque requête et garde un seul pool de connexions pour le processus. Nommez le module comme vous voulez, sauf openemail.py, qui peut masquer le paquet.

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)

Le client openemail fourni fonctionne aussi ici : appelez init(settings.OPENEMAIL_API_KEY) une fois, depuis la méthode ready() de la configuration de votre application, et importez openemail partout où vous envoyez.

Flask

La fabrique d'application construit le client en même temps que l'application et le garde dans app.extensions, et les vues y accèdent via current_app. Une fabrique qui accepte aussi un client permet à un test d'en passer un construit sur 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

Construisez un seul AsyncOpenEmail dans le lifespan, pour qu'il appartienne à la boucle d'événements qui sert les requêtes et se ferme à l'arrêt du serveur, et transmettez-le aux routes au moyen d'une dépendance.

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']}

Un test remplace le client via app.dependency_overrides[get_openemail] : aucune route n'atteint donc l'API.

Endpoints de webhook

Vérifiez chaque livraison avant d'agir dessus, avec le corps brut et les en-têtes de la requête : await request.body() et request.headers dans FastAPI, request.body et request.headers dans Django, et request.get_data() et request.headers dans Flask. La recherche d'en-tête ignore la casse : l'objet d'en-têtes propre à chaque framework fonctionne donc tel quel.

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 refuse un POST qui ne porte pas de jeton CSRF, et une livraison n'en porte pas : la vue est donc csrf_exempt, et c'est la signature qui prouve que la requête vient d'OpenEmail. Lisez request.headers plutôt que request.META, dont les clés sont renommées sous la forme HTTP_X_OPENEMAIL_SIGNATURE.

Répondez vite par un 2xx et faites le travail ensuite. Une livraison qui ne reçoit aucune réponse, ou qui reçoit un 408, 425, 429 ou 5xx, est retentée jusqu'à 8 fois en 27 heures et demie environ, et un rejeu renvoie un événement avec le même id : conservez donc les id que vous avez traités et ignorez une répétition.

Vérifier une livraisonCe que rejette la vérification de signature, et pourquoi.Points de terminaisonEnregistrer un endpoint, le tester et rejouer une livraison.

Tâches en arrière-plan

Une file de tâches relance une tâche quand elle échoue, et une tâche peut échouer après le départ de son e-mail : la réponse s'est perdue, ou le worker s'est arrêté avant d'avoir terminé. Passez un idempotency_key= dérivé de ce qui a rendu l'envoi nécessaire. Chaque exécution de la tâche porte alors la même clé, si bien qu'une répétition rejoue le message d'origine au lieu d'en envoyer un second.

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']

Avec acks_late=True, Celery n'acquitte une tâche qu'une fois exécutée : une tâche interrompue par un worker qui s'est arrêté peut donc être relivrée, ce qui est sans danger ici, car la clé transforme cette seconde exécution en rejeu. Retry de RQ relance un job échoué, avec le même effet.

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]))

Dérivez la clé de ce qui a rendu l'envoi nécessaire, jamais d'une horloge. Elle compte de 1 à 255 lettres, chiffres, traits de soulignement, points, deux-points ou traits d'union : construisez-la donc à partir d'un id plutôt que d'une adresse e-mail. Construisez aussi le corps à partir des seuls arguments de la tâche : une répétition avec la même clé et un corps différent est refusée avec 422 idempotency_key_reuse au lieu d'être rejouée.

Construisez le client au niveau du module. Il n'ouvre aucune connexion avant sa première requête : chaque processus worker forké depuis le parent ouvre donc les siennes.

Réessais et idempotenceCe que le client réessaie, et pourquoi un envoi réessayé ne peut pas se répéter.