Фреймворки
Django, Flask и FastAPI, эндпоинт для вебхуков и фоновые задачи, которые никогда не отправляют дважды.
Django
Храните ключ в настройках, читая его из окружения, и постройте один OpenEmail в отдельном модуле, который импортируют представления. Клиент можно безопасно использовать из нескольких потоков, так что один экземпляр обслуживает все запросы и держит один пул соединений на процесс. Назовите модуль как угодно, только не openemail.py, иначе он может заслонить пакет.
import os OPENEMAIL_API_KEY = os.environ['OPENEMAIL_API_KEY']OPENEMAIL_TIMEOUT = 10.0OPENEMAIL_SENDER = 'Acme <[email protected]>'from django.conf import settingsfrom openemail import OpenEmail mailer = OpenEmail(settings.OPENEMAIL_API_KEY, timeout=settings.OPENEMAIL_TIMEOUT)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)Поставляемый клиент openemail здесь тоже работает: вызовите init(settings.OPENEMAIL_API_KEY) один раз из метода ready() конфигурации вашего приложения и импортируйте openemail везде, где отправляете.
Flask
Фабрика приложения строит клиент вместе с приложением и хранит его в app.extensions, а представления получают его через current_app. Фабрика, которая также принимает клиент, позволяет тесту передать клиент, построенный на httpx.MockTransport.
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 appFastAPI
Постройте один AsyncOpenEmail в lifespan, чтобы он принадлежал циклу событий, который обслуживает запросы, и закрывался при остановке сервера, и передавайте его маршрутам через зависимость.
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']}Тест подменяет клиент через app.dependency_overrides[get_openemail], так что ни один маршрут не обращается к API.
Эндпоинты вебхуков
Проверяйте каждую доставку, прежде чем на неё реагировать, по сырому телу и заголовкам запроса: await request.body() и request.headers в FastAPI, request.body и request.headers в Django, request.get_data() и request.headers в Flask. Поиск заголовка не учитывает регистр, поэтому собственный объект заголовков каждого фреймворка работает как есть.
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)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 отклоняет POST без CSRF-токена, а у доставки его нет, поэтому представление помечено csrf_exempt: то, что запрос пришёл от OpenEmail, доказывает подпись. Читайте request.headers, а не request.META, ключи которого переименованы в форму HTTP_X_OPENEMAIL_SIGNATURE.
Отвечайте 2xx быстро, а работу делайте потом. Доставка, на которую не пришёл ответ или пришёл 408, 425, 429 или 5xx, повторяется до 8 раз примерно за 27 с половиной часов, а повторная отправка присылает событие снова с тем же id, поэтому сохраняйте обработанные идентификаторы и пропускайте повторы.
Фоновые задачи
Очередь задач перезапускает задачу, когда та падает, а задача может упасть уже после того, как её письмо ушло: ответ потерялся или воркер остановился, не закончив. Передайте idempotency_key=, выведенный из того, что сделало отправку необходимой. Тогда каждый запуск задачи несёт один и тот же ключ, и повтор воспроизводит исходное письмо, а не отправляет второе.
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']С acks_late=True Celery подтверждает задачу только после её выполнения, так что задача, прерванная остановившимся воркером, может быть доставлена снова, и здесь это безопасно, потому что ключ превращает второй запуск в воспроизведение. Retry из RQ перезапускает упавшее задание с тем же эффектом.
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]))Выводите ключ из того, что сделало отправку необходимой, и никогда из часов. Он содержит от 1 до 255 букв, цифр, подчёркиваний, точек, двоеточий или дефисов, поэтому стройте его из идентификатора, а не из адреса электронной почты. Тело тоже стройте только из аргументов задачи: повтор с тем же ключом и другим телом отклоняется с 422 idempotency_key_reuse, а не воспроизводится.
Стройте клиент на уровне модуля. Он не открывает соединений до первого запроса, поэтому каждый рабочий процесс, порождённый от родительского через fork, открывает свои.