Python

Framework'ler

Django, Flask ve FastAPI, bir webhook uç noktası ve asla iki kez göndermeyen arka plan işleri.

Django

Anahtarı ortamdan okunmuş olarak settings içinde tutun ve view'ların içe aktardığı ayrı bir modülde tek bir OpenEmail oluşturun. İstemciyi iş parçacıkları arasında paylaşmak güvenlidir; bu yüzden tek bir örnek her isteğe hizmet eder ve süreç için tek bir bağlantı havuzu tutar. Modüle openemail.py dışında herhangi bir ad verin, çünkü bu ad paketi gölgeleyebilir.

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)

Paketle gelen openemail istemcisi burada da çalışır: init(settings.OPENEMAIL_API_KEY) çağrısını uygulama yapılandırmanızın ready() metodundan bir kez yapın ve gönderim yaptığınız her yerde openemail'i içe aktarın.

Flask

Uygulama fabrikası istemciyi uygulamayla birlikte oluşturur ve app.extensions içinde tutar; view'lar ona current_app üzerinden ulaşır. Bir istemciyi de kabul eden bir fabrika, bir testin httpx.MockTransport üzerine kurulu bir istemci geçirmesine olanak tanır.

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

Lifespan içinde tek bir AsyncOpenEmail oluşturun; böylece istekleri karşılayan olay döngüsüne ait olur ve sunucu durduğunda kapanır. Ardından onu bir bağımlılıkla route'lara verin.

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

Bir test, istemciyi app.dependency_overrides[get_openemail] üzerinden değiştirir; böylece hiçbir route API'ye ulaşmaz.

Webhook uç noktaları

Her teslimatı, üzerine işlem yapmadan önce ham gövde ve istek başlıklarıyla doğrulayın: FastAPI'de await request.body() ve request.headers, Django'da request.body ve request.headers, Flask'ta ise request.get_data() ve request.headers. Başlık araması büyük/küçük harfe duyarsızdır; bu yüzden her framework'ün kendi başlık nesnesi olduğu gibi çalışır.

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, CSRF token'ı taşımayan bir POST isteğini reddeder ve bir teslimat hiç token taşımaz; bu yüzden view csrf_exempt olur: isteğin OpenEmail'den geldiğini kanıtlayan şey imzadır. request.META yerine request.headers okuyun, çünkü onun anahtarları HTTP_X_OPENEMAIL_SIGNATURE biçimine dönüştürülür.

Hızlıca bir 2xx ile yanıt verin ve işi sonra yapın. Yanıt almayan ya da 408, 425, 429 veya 5xx alan bir teslimat, yaklaşık 27 buçuk saat içinde en fazla 8 kez yeniden denenir ve bir yeniden oynatma bir olayı aynı id ile yeniden gönderir; bu yüzden işlediğiniz id'leri saklayın ve tekrarları atlayın.

Bir teslimatı doğrulamakİmza denetiminin neleri reddettiği ve neden.Uç noktalarBir uç nokta kaydedin, test edin ve bir teslimatı yeniden oynatın.

Arka plan işleri

Bir iş kuyruğu, başarısız olan bir görevi yeniden çalıştırır ve bir görev, e-postası çıktıktan sonra da başarısız olabilir: yanıt kaybolmuştur ya da worker işi bitirmeden durmuştur. Gönderimi gerekli kılan şeyden türetilmiş bir idempotency_key= geçirin. Böylece görevin her çalışması aynı anahtarı taşır ve bir tekrar, ikinci bir mesaj göndermek yerine özgün mesajı yeniden oynatır.

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

acks_late=True ile Celery bir görevi ancak çalıştıktan sonra onaylar; bu yüzden duran bir worker yüzünden yarıda kalan bir görev yeniden teslim edilebilir. Bu burada güvenlidir, çünkü anahtar o ikinci çalışmayı bir yeniden oynatmaya dönüştürür. RQ'nun Retry mekanizması başarısız bir işi aynı etkiyle yeniden çalıştırır.

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

Anahtarı gönderimi gerekli kılan şeyden türetin, asla bir saatten değil. 1 ila 255 harf, rakam, alt çizgi, nokta, iki nokta üst üste ya da kısa çizgiden oluşur; bu yüzden onu bir e-posta adresinden değil bir id'den oluşturun. Gövdeyi de yalnızca görevin argümanlarından oluşturun: aynı anahtarla ve farklı bir gövdeyle yapılan bir tekrar yeniden oynatılmaz, 422 idempotency_key_reuse ile reddedilir.

İstemciyi modül düzeyinde oluşturun. İlk isteğine kadar hiçbir bağlantı açmaz; bu yüzden üst süreçten fork edilen her worker süreci kendi bağlantısını açar.

Yeniden denemeler ve idempotencyİstemcinin neleri yeniden denediği ve yeniden denenen bir gönderimin neden tekrarlanamayacağı.