---
title: "Planning a transactional email send from FastAPI"
canonical: https://pulsepigeon.com/blog/send-email-with-fastapi
last_updated: 2026-10-20
---

# 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.

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](/docs/sdks) and [developer API reference](/docs/developer) 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"]
```