Send Emails with FastAPI
Learn how to send transactional emails using PostStack and FastAPI.
FastAPI is async-first, so reach for the `AsyncPostStack` client and `await` it directly inside your route handlers — the SDK uses httpx under the hood and shares a connection pool, which keeps tail latency low under load. Create the client once in a `lifespan` context manager, expose it through a `Depends` dependency, and you get clean injection plus a single pool for the whole app. For fire-and-forget notifications, `BackgroundTasks` is enough at low volume; for high throughput, hand off to arq or Celery.
1. Install the SDK
pip install poststack2. Initialize the client
# main.py — create one AsyncPostStack for the app lifespan
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI
from poststack import AsyncPostStack
@asynccontextmanager
async def lifespan(app: FastAPI):
# The async client is itself an async context manager, so open it for the
# app's lifetime and let it clean up its connection pool on shutdown.
async with AsyncPostStack(api_key=os.environ["POSTSTACK_API_KEY"]) as ps:
app.state.poststack = ps
yield
app = FastAPI(lifespan=lifespan)
# dependency — inject the shared client into route handlers
from fastapi import Request
def get_poststack(request: Request) -> AsyncPostStack:
return request.app.state.poststack3. Send an email
from fastapi import Depends, FastAPI
from pydantic import BaseModel, EmailStr
from poststack import AsyncPostStack
class SignupBody(BaseModel):
email: EmailStr
name: str
@app.post("/signup")
async def signup(
body: SignupBody,
ps: AsyncPostStack = Depends(get_poststack),
):
user = await create_user(body)
# await the async client directly inside the async handler
result = await ps.emails.send({
"from": "hello@yourdomain.com",
"to": [body.email],
"subject": "Welcome to Acme",
"html": f"<h1>Welcome, {body.name}!</h1>",
})
return {"ok": True, "email_id": result["id"]}
# Fire-and-forget with BackgroundTasks (low volume)
from fastapi import BackgroundTasks
@app.post("/notify")
async def notify(body: SignupBody, bg: BackgroundTasks, ps: AsyncPostStack = Depends(get_poststack)):
bg.add_task(ps.emails.send, {
"from": "hello@yourdomain.com",
"to": [body.email],
"subject": "Heads up",
"html": "<p>Something happened.</p>",
})
return {"queued": True}4. Handle errors
FastAPI idioms for error handling, retries, and structured logging when calling the PostStack API.
import logging
from poststack import AsyncPostStack, PostStackError
logger = logging.getLogger(__name__)
async def send(ps: AsyncPostStack, payload):
# The async client retries 429/5xx internally; handle terminal errors here.
try:
return await ps.emails.send(payload)
except PostStackError as e:
if 400 <= e.status_code < 500 and e.status_code != 429:
logger.error("PostStack %s: %s (req %s)", e.code, e.error, e.request_id)
return None
raiseFramework integrations
Lifespan + dependency injection
Construct `AsyncPostStack` in an `@asynccontextmanager` lifespan, store it on `app.state`, and expose a `get_poststack` dependency. Handlers receive the shared client via `Depends`, so there is exactly one HTTP connection pool for the process.
BackgroundTasks for fire-and-forget
For low-volume notifications, `bg.add_task(ps.emails.send, payload)` returns the response before the email is sent. Note BackgroundTasks run in-process after the response — they do not survive a crash, so use a real queue for anything you must not lose.
arq / Celery for high volume
When you send at scale or need durable retries, enqueue to arq (async-native) or Celery instead of BackgroundTasks. The worker constructs its own `AsyncPostStack` and awaits the send; the SDK retries transient API errors itself, so the worker only re-queues on a terminal `PostStackError`.
Pydantic request models
Validate recipient addresses with `EmailStr` on your request model so malformed input is rejected before it ever reaches the API — fewer 4xx errors and cleaner suppression lists.
Common pitfalls
Using the sync client in an async handler
Calling the synchronous `PostStack` client inside `async def` blocks the event loop and collapses throughput. Always use `AsyncPostStack` in async code, or wrap sync calls in `asyncio.to_thread`.
Forgetting to await the send
`await ps.emails.send(...)` — without `await` you get a coroutine object and no email is ever sent. Enable a type checker (mypy/pyright) to catch the missing await.
Relying on BackgroundTasks for critical mail
BackgroundTasks run in-process and are lost if the worker restarts mid-task. For receipts and password resets, use a durable queue (arq/Celery), not BackgroundTasks.
Notes
- Use `AsyncPostStack` and `await` it directly inside `async def` handlers
- Create the client in `lifespan` and inject it with `Depends` so one connection pool is shared
FAQ
How do I send email in a FastAPI route?
Use `AsyncPostStack`, inject it with `Depends`, and `await ps.emails.send(...)` directly in your `async def` handler. Create the client once in a `lifespan` context manager so the whole app shares one connection pool.
Should I use BackgroundTasks or a queue?
BackgroundTasks is fine for low-volume, non-critical notifications — it runs after the response is sent. For mail you must not lose (receipts, password resets) or high volume, use a durable queue like arq or Celery.
Can I use the synchronous client in FastAPI?
You can, but only if you wrap it in `asyncio.to_thread` — calling the sync client directly in an `async def` blocks the event loop. The async client (`AsyncPostStack`) is the right default for FastAPI.
Related guides
Ready to send emails with FastAPI?
Create a free account and get your API key in under a minute.