--- title: "Frameworks" description: "Django, Flask and FastAPI, a webhook endpoint, and background jobs that never send twice." url: "https://openemail.uk/docs/python/frameworks" area: "Python" category: "Getting started" --- # 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.0 OPENEMAIL_SENDER = 'Acme ' ``` **mailer.py** ``` from django.conf import settings from openemail import OpenEmail mailer = OpenEmail(settings.OPENEMAIL_API_KEY, timeout=settings.OPENEMAIL_TIMEOUT) ``` **views.py** ``` from django.conf import settings from django.http import HttpRequest, JsonResponse from django.views.decorators.http import require_POST from openemail import OpenEmailApiError from acme.mailer import mailer @require_POST def 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, request from 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 ', '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 AsyncIterator from contextlib import asynccontextmanager from typing import Annotated from fastapi import Body, Depends, FastAPI, Request from openemail import AsyncOpenEmail @asynccontextmanager async 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 ', '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, Response from 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, HttpResponse from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_POST from openemail import WebhookVerificationError, verify_webhook_signature from acme.models import ReceivedEvent @csrf_exempt @require_POST def 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 delivery](https://openemail.uk/docs/python/webhooks/verify.md): What the signature check rejects, and why. - [Endpoints](https://openemail.uk/docs/python/webhooks/endpoints.md): Register 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_task from 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 ', '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 Redis from 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 ', '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', 'ada@example.com', 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 idempotency](https://openemail.uk/docs/python/retries.md): What the client retries, and why a retried send cannot repeat.