Send email from FastAPI without blocking the loop.

1. Create the client once and inject it.

   The lifespan handler builds one client on app.state before the first request. get_notix reads it back, and the NotixDep alias makes every handler that needs it one parameter long.

   acme/main.py

   # acme/main.py
   import os
   from contextlib import asynccontextmanager
   from typing import Annotated
   
   from fastapi import BackgroundTasks, Depends, FastAPI, HTTPException, Request
   from pydantic import BaseModel, EmailStr
   from notix import Notix
   from notix.webhooks import WebhookVerificationError
   
   
   @asynccontextmanager
   async def lifespan(app: FastAPI):
       # One client for the process; it keeps a requests.Session for connection reuse.
       # raise_on_error=False returns (None, error) instead of raising.
       app.state.notix = Notix(os.environ["NOTIX_API_KEY"], raise_on_error=False)
       # Pointing at a different Notix base URL (a staging environment, say)? Pass url=.
       yield
   
   
   app = FastAPI(lifespan=lifespan)
   
   
   def get_notix(request: Request) -> Notix:
       return request.app.state.notix
   
   
   NotixDep = Annotated[Notix, Depends(get_notix)]
   

2. Send from a route.

   The package is synchronous, so keep it off the event loop: hand the send to BackgroundTasks and answer 202, or declare the handler with plain def so FastAPI runs it in the threadpool and the response can carry the emailId. Either way the idempotency key comes from the business event, never from a random value.

   Background taskWait for the id

   acme/main.py

   class ReceiptRequest(BaseModel):
       email: EmailStr
       total: str
   
   
   def send_receipt(notix: Notix, order_id: int, body: ReceiptRequest) -> None:
       # A plain function passed to add_task; it runs after the response is sent.
       data, err = notix.emails.send(
           {
               "from": "Acme <receipts@acme.com>",
               "to": body.email,
               "templateId": "tpl_receipt_v4",
               "variables": {"orderId": str(order_id), "total": body.total},
           },
           # One key per business event: a retried request can never send twice.
           options={"idempotency_key": f"order-{order_id}-receipt"},
       )
       if err:
           # Log it; the client already got 202. Retrying later is safe because of the key.
           log.warning("receipt %s failed: %s %s", order_id, err["code"], err["message"])
           return
       store_email_id(order_id, data["emailId"])   # the key on every webhook event
   
   
   @app.post("/orders/{order_id}/receipt", status_code=202)
   async def queue_receipt(
       order_id: int, body: ReceiptRequest, notix: NotixDep, background: BackgroundTasks
   ):
       background.add_task(send_receipt, notix, order_id, body)
       return {"queued": True}
   

3. Receive delivery events.

   Delivered, bounced and complained arrive later as signed email.* events. Verify with the raw bytes from await request.body(); a re-serialised model no longer matches the signature.

   acme/main.py

   # acme/main.py
   @app.post("/webhooks/notix")
   async def notix_webhook(request: Request, notix: NotixDep):
       webhooks = notix.webhooks(os.environ["NOTIX_WEBHOOK_SECRET"])
       try:
           # await request.body() is the raw bytes; the signature covers exactly those.
           event = webhooks.construct_event(await request.body(), headers=dict(request.headers))
       except WebhookVerificationError as error:
           raise HTTPException(status_code=400, detail=str(error))
   
       email_id = event["data"]["id"]
       if event["type"] == "email.delivered":
           mark_receipt(email_id, "delivered")
       elif event["type"] == "email.bounced" and event["data"]["bounce"]["type"] == "Permanent":
           mark_receipt(email_id, "bounced")   # the address is now suppressed by Notix
   
       return {"ok": True}
   

4. Prove it without the network.

   Swap the client on app.state for a mock and assert the payload and key that reached it. TestClient runs background tasks before it returns, so the assertion sees the call.

   tests/test_receipt.py

   # tests/test_receipt.py
   from unittest.mock import MagicMock
   
   from fastapi.testclient import TestClient
   
   from acme.main import app
   
   
   def test_receipt_is_queued_with_an_idempotency_key():
       fake = MagicMock()
       fake.emails.send.return_value = ({"emailId": "eml_1"}, None)
       app.state.notix = fake
   
       with TestClient(app) as client:           # runs lifespan; state set above survives
           response = client.post("/orders/4471/receipt", json={"email": "a@example.com", "total": "₦12,400"})
   
       assert response.status_code == 202
       # TestClient runs background tasks before returning.
       _, kwargs = fake.emails.send.call_args
       assert kwargs["options"] == {"idempotency_key": "order-4471-receipt"}
   

Register the webhook URL in the dashboard and copy its secret into NOTIX_WEBHOOK_SECRET. The scheme is on the webhook signature page.