PulsePigeonemail infrastructure

Blog

Planning a transactional email send from FastAPI

Conceptual guidance for server-side sends from FastAPI, including idempotency, response handling, and when a durable worker is safer than BackgroundTasks.

By Administrator · 2026-08-26 · updated 2026-10-20 · 1 min read

FastAPI's BackgroundTasks runs work after the response, but it is in-process and does not provide durable delivery or automatic retries. Use it only when that loss model is acceptable; use a durable queue for critical password-reset, verification, or billing mail.

Pass an idempotency key derived from the triggering business event. For a password reset, reuse the persisted reset-request ID; for an order confirmation, reuse the order ID. A UUID minted per inbound request can change when a client or upstream retry re-enters the route, which can enqueue a duplicate.

Inside an async def background task, use httpx (or another async HTTP client) to call PulsePigeon's REST API so synchronous network I/O does not block the event loop. A normal def background task may use the synchronous requests library because Starlette runs that task in its thread pool, but thread-pool capacity is still finite.

The published pulsepigeon Python package provides synchronous and asynchronous clients. Install version 1.0.0 from PyPI with pip install pulsepigeon for the project-key surface; your worker remains responsible for bounded retries. Consult the SDK method reference and developer API reference before wiring the application request.

Record the accepted send response's id, not the API key or message content, in your application trace. Use that value to query tenant-scoped logs or events and to correlate signed delivery-status webhook payloads, whose message_id is the same identifier.

Send helper with the Python SDK: send_reset_email.py

import os

from pulsepigeon_client import (
    PulsePigeonClient,
    PulsePigeonRateLimitError,
    PulsePigeonServerError,
)

client = PulsePigeonClient(api_key=os.environ["PULSEPIGEON_API_KEY"])


class RetryLater(Exception):
    def __init__(self, after_seconds: int | None) -> None:
        super().__init__("retry later")
        self.after_seconds = after_seconds


def send_reset_email(reset_request_id: str, project_id: str, to: str, reset_url: str) -> str:
    # Call from a durable worker, or from a plain def background task if losing the send is acceptable.
    try:
        accepted = client.send_message(
            {
                "project_id": project_id,
                "from_email": os.environ["PULSEPIGEON_FROM_EMAIL"],
                "to": [{"email": to}],
                "subject": "Reset your password",
                "category": "security",
                "text": f"Use this link to reset your password: {reset_url}",
            },
            idempotency_key=f"reset-{reset_request_id}",
        )
    except PulsePigeonRateLimitError as exc:
        # Raise so the worker's retry policy runs; returning marks the job done.
        raise RetryLater(exc.retry_after_seconds) from exc
    except PulsePigeonServerError as exc:
        raise RetryLater(None) from exc
    return accepted["id"]