High-throughput async email dispatch in Python
Using FastAPI with httpx allows Python backends to handle thousands of concurrent email requests without blocking worker threads.
Pydantic v2 compiles validation rules directly into Rust, guaranteeing sub-millisecond payload validation before hitting outbound email gateways.
FastAPI Service Implementation (main.py)
Define the async endpoint with Pydantic v2 models and connection pooling.
PYTHON
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, EmailStr, Field
import httpx
import os
app = FastAPI(title="SadaSend Email Microservice", version="1.0.0")
class EmailPayload(BaseModel):
to: EmailStr = Field(description="Recipient email address")
subject: str = Field(min_length=1, max_length=200)
text_content: str = Field(min_length=1)
idempotency_key: str | None = None
@app.post("/api/v1/send", status_code=status.HTTP_202_ACCEPTED)
async def dispatch_email(payload: EmailPayload):
api_key = os.getenv("SADASEND_API_KEY")
if not api_key:
raise HTTPException(status_code=500, detail="SADASEND_API_KEY unconfigured")
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
}
if payload.idempotency_key:
headers["Idempotency-Key"] = payload.idempotency_key
async with httpx.AsyncClient(timeout=5.0) as client:
res = await client.post(
"https://api.sadasend.com/v1/emails",
headers=headers,
json={"to": payload.to, "subject": payload.subject, "text": payload.text_content}
)
if res.status_code == 403:
raise HTTPException(status_code=403, detail="Recipient domain not in authorized allowlist")
if not res.is_success:
raise HTTPException(status_code=res.status_code, detail=res.text)
return res.json()Production tips for FastAPI email microservices
- Reuse a global httpx.AsyncClient across requests to take advantage of HTTP keep-alive connection pooling.
- Always supply Idempotency-Key headers from client payloads to prevent duplicate sends on network retries.
Early Access
Building AI agents that send email?
Join the SadaSend early access waitlist to get scoped API keys, recipient allowlists, and Model Context Protocol (MCP) servers upon launch.