Python

Frameworks

Django, Flask e FastAPI, um endpoint de webhooks e tarefas em segundo plano que nunca enviam duas vezes.

Django

Guarde a chave nas definições, lida a partir do ambiente, e construa um único OpenEmail num módulo próprio que as vistas importem. É seguro partilhar o cliente entre threads, por isso uma única instância serve todos os pedidos e mantém um único pool de ligações para o processo. Dê ao módulo qualquer nome exceto openemail.py, que pode esconder o pacote.

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)

O cliente openemail incluído também funciona aqui: chame init(settings.OPENEMAIL_API_KEY) uma vez, a partir do método ready() da configuração da sua aplicação, e importe openemail onde quer que envie.

Flask

A fábrica de aplicações constrói o cliente com a aplicação e guarda-o em app.extensions, e as vistas chegam a ele através de current_app. Uma fábrica que também aceite um cliente permite que um teste passe um construído 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

Construa um único AsyncOpenEmail no lifespan, para que pertença ao ciclo de eventos que serve os pedidos e seja fechado quando o servidor parar, e passe-o às rotas com uma dependência.

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

Um teste substitui o cliente através de app.dependency_overrides[get_openemail], por isso nenhuma rota chega à API.

Endpoints de webhook

Verifique cada entrega antes de agir sobre ela, com o corpo em bruto e os cabeçalhos do pedido: await request.body() e request.headers no FastAPI, request.body e request.headers no Django, e request.get_data() e request.headers no Flask. A procura do cabeçalho ignora maiúsculas e minúsculas, por isso o objeto de cabeçalhos de cada framework funciona tal como está.

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)

O Django recusa um POST que não traga token CSRF, e uma entrega não traz nenhum, por isso a vista é csrf_exempt: a assinatura é o que prova que o pedido veio da OpenEmail. Leia request.headers em vez de request.META, cujas chaves são renomeadas para a forma HTTP_X_OPENEMAIL_SIGNATURE.

Responda depressa com um 2xx e faça o trabalho depois. Uma entrega que não recebe resposta, ou que recebe um 408, 425, 429 ou 5xx, é tentada de novo, até 8 vezes em cerca de 27 horas e meia, e um reenvio manda outra vez um evento com o mesmo id, por isso guarde os ids que já tratou e ignore as repetições.

Verificar uma entregaO que a verificação da assinatura rejeita, e porquê.EndpointsRegistar um endpoint, testá-lo e reenviar uma entrega.

Tarefas em segundo plano

Uma fila de tarefas volta a executar uma tarefa quando ela falha, e uma tarefa pode falhar depois de o seu email ter saído: a resposta perdeu-se, ou o worker parou antes de terminar. Passe um idempotency_key= derivado do que tornou o envio necessário. Assim, cada execução da tarefa leva a mesma chave, e uma repetição reproduz a mensagem original em vez de enviar uma segunda.

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

Com acks_late=True, o Celery só confirma uma tarefa depois de ela ter sido executada, por isso uma tarefa interrompida por um worker que parou pode ser entregue de novo, o que aqui é seguro porque a chave transforma essa segunda execução numa reprodução. O Retry do RQ volta a executar uma tarefa falhada, com o mesmo efeito.

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 a chave do que tornou o envio necessário, nunca de um relógio. Tem de 1 a 255 letras, dígitos, sublinhados, pontos, dois pontos ou hífenes, por isso construa-a a partir de um id e não de um endereço de email. Construa também o corpo apenas a partir dos argumentos da tarefa: uma repetição com a mesma chave e um corpo diferente é recusada com 422 idempotency_key_reuse em vez de ser reproduzida.

Construa o cliente ao nível do módulo. Não abre nenhuma ligação até ao primeiro pedido, por isso cada processo worker criado por fork a partir do pai abre a sua.

Repetições e idempotênciaO que o cliente volta a tentar, e porque é que um envio repetido não pode sair duas vezes.