Python

Frameworks

Django, Flask and FastAPI, a webhook endpoint, and background jobs that never send twice.

Django

Keep the key in settings, read from the environment, and build one OpenEmail in a module of its own that views import. The client is safe to share between threads, so one instance serves every request and keeps one connection pool for the process. Name the module anything but openemail.py, which can hide the package.

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)

The shipped openemail client works here too: call init(settings.OPENEMAIL_API_KEY) once, from the ready() method of your app config, and import openemail wherever you send.

Flask

The app factory builds the client with the app and keeps it in app.extensions, and views reach it through current_app. A factory that also accepts a client lets a test pass one built on 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

Build one AsyncOpenEmail in the lifespan, so it belongs to the event loop that serves requests and closes when the server stops, and hand it to routes with a dependency.

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

A test swaps the client through app.dependency_overrides[get_openemail], so no route reaches the API.

Webhook endpoints

Verify every delivery before acting on it, with the raw body and the request headers: await request.body() and request.headers in FastAPI, request.body and request.headers in Django, and request.get_data() and request.headers in Flask. The header lookup ignores case, so each framework’s own headers object works as it is.

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 refuses a POST that carries no CSRF token, and a delivery carries none, so the view is csrf_exempt: the signature is what proves the request came from OpenEmail. Read request.headers rather than request.META, whose keys are renamed to the HTTP_X_OPENEMAIL_SIGNATURE form.

Answer with a 2xx quickly and do the work afterwards. A delivery that gets no answer, or a 408, 425, 429 or 5xx, is tried again, up to 8 times in about 27 and a half hours, and a replay sends an event again with the same id, so keep the ids you have handled and skip a repeat.

Verifying a deliveryWhat the signature check rejects, and why.EndpointsRegister an endpoint, test it and replay a delivery.

Background jobs

A job queue runs a task again when it fails, and a task can fail after its email has left: the response was lost, or the worker stopped before it finished. Pass an idempotency_key= derived from what made the send necessary. Every run of the task then carries the same key, so a repeat replays the original message instead of sending a second one.

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

With acks_late=True, Celery acknowledges a task only once it has run, so a task cut short by a worker that stopped can be delivered again, which is safe here because the key turns that second run into a replay. RQ’s Retry runs a failed job again, with the same effect.

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

Derive the key from what made the send necessary, never from a clock. It is 1 to 255 letters, digits, underscores, dots, colons or hyphens, so build it from an id rather than an email address. Build the body from the task’s arguments alone as well: a repeat with the same key and a different body is refused with 422 idempotency_key_reuse rather than replayed.

Build the client at module level. It opens no connection until its first request, so each worker process forked from the parent opens its own.

Retries and idempotencyWhat the client retries, and why a retried send cannot repeat.