Python

Frameworks

Django, Flask und FastAPI, ein Webhook-Endpunkt und Hintergrundjobs, die nie doppelt senden.

Django

Legen Sie den Schlüssel in den Settings ab, aus der Umgebung gelesen, und erzeugen Sie einen OpenEmail in einem eigenen Modul, das die Views importieren. Der Client lässt sich gefahrlos zwischen Threads teilen, eine Instanz bedient also jede Anfrage und hält einen Verbindungspool für den Prozess. Nennen Sie das Modul beliebig, nur nicht openemail.py, denn das kann das Paket verdecken.

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)

Der mitgelieferte openemail-Client funktioniert hier ebenfalls: Rufen Sie init(settings.OPENEMAIL_API_KEY) einmal auf, aus der Methode ready() Ihrer App-Konfiguration, und importieren Sie openemail überall dort, wo Sie senden.

Flask

Die App-Factory erzeugt den Client zusammen mit der App und legt ihn in app.extensions ab, und Views erreichen ihn über current_app. Eine Factory, die zusätzlich einen Client annimmt, erlaubt einem Test, einen auf httpx.MockTransport gebauten zu übergeben.

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

Erzeugen Sie einen AsyncOpenEmail im Lifespan, damit er zu der Event-Loop gehört, die die Anfragen bedient, und beim Stoppen des Servers geschlossen wird, und reichen Sie ihn über eine Dependency an die Routen weiter.

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

Ein Test tauscht den Client über app.dependency_overrides[get_openemail] aus, sodass keine Route die API erreicht.

Webhook-Endpunkte

Verifizieren Sie jede Zustellung, bevor Sie darauf reagieren, mit dem rohen Body und den Request-Headern: await request.body() und request.headers in FastAPI, request.body und request.headers in Django sowie request.get_data() und request.headers in Flask. Die Header-Suche ignoriert Groß- und Kleinschreibung, das Header-Objekt jedes Frameworks funktioniert also so, wie es ist.

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 lehnt ein POST ohne CSRF-Token ab, und eine Zustellung trägt keines, die View ist daher csrf_exempt: Die Signatur ist das, was beweist, dass die Anfrage von OpenEmail kam. Lesen Sie request.headers statt request.META, dessen Schlüssel in die Form HTTP_X_OPENEMAIL_SIGNATURE umbenannt werden.

Antworten Sie schnell mit einem 2xx und erledigen Sie die Arbeit danach. Eine Zustellung, die keine Antwort bekommt oder einen 408, 425, 429 oder 5xx, wird erneut versucht, bis zu 8-mal, verteilt über rund 27 und eine halbe Stunde, und ein Replay sendet ein Event erneut mit derselben id. Merken Sie sich also die IDs, die Sie verarbeitet haben, und überspringen Sie eine Wiederholung.

Eine Zustellung verifizierenWas die Signaturprüfung ablehnt und warum.EndpunkteEinen Endpunkt registrieren, testen und eine Zustellung erneut senden.

Hintergrundjobs

Eine Job-Queue führt einen Task erneut aus, wenn er fehlschlägt, und ein Task kann fehlschlagen, nachdem seine E-Mail bereits hinaus ist: Die Antwort ging verloren, oder der Worker hat vor dem Ende angehalten. Übergeben Sie einen idempotency_key=, abgeleitet aus dem, was den Versand nötig gemacht hat. Jeder Lauf des Tasks trägt dann denselben Schlüssel, sodass eine Wiederholung die ursprüngliche Nachricht erneut abspielt, statt eine zweite zu senden.

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

Mit acks_late=True bestätigt Celery einen Task erst, nachdem er gelaufen ist, ein Task, der durch einen angehaltenen Worker abgebrochen wurde, kann also erneut zugestellt werden. Das ist hier unbedenklich, weil der Schlüssel diesen zweiten Lauf zu einem erneuten Abspielen macht. Retry von RQ führt einen fehlgeschlagenen Job erneut aus, mit derselben Wirkung.

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

Leiten Sie den Schlüssel aus dem ab, was den Versand nötig gemacht hat, nie aus einer Uhr. Er besteht aus 1 bis 255 Buchstaben, Ziffern, Unterstrichen, Punkten, Doppelpunkten oder Bindestrichen, bauen Sie ihn daher aus einer ID statt aus einer E-Mail-Adresse. Bauen Sie auch den Body allein aus den Argumenten des Tasks: Eine Wiederholung mit demselben Schlüssel und einem anderen Body wird mit 422 idempotency_key_reuse abgelehnt, statt erneut abgespielt zu werden.

Erzeugen Sie den Client auf Modulebene. Er öffnet bis zu seiner ersten Anfrage keine Verbindung, sodass jeder vom Elternprozess abgezweigte Worker-Prozess seine eigene öffnet.

Wiederholungsversuche und IdempotenzWas der Client wiederholt und warum ein wiederholter Versand nicht doppelt hinausgehen kann.