FastAPI
Send transactional emails from your FastAPI application using the Veil Mail Python SDK.
Prerequisites
- FastAPI 0.100 or later
- Python 3.9+
- A Veil Mail API key
- A verified domain
1. Install the SDK
pip install veilmail2. Configure the Client
Create a shared client instance. Store your API key in an environment variable — never commit it to source control.
.env
VEILMAIL_API_KEY=veil_live_xxxxx
VEILMAIL_WEBHOOK_SECRET=whsec_xxxxxapp/veilmail_client.py
import os
from veilmail import VeilMail
veilmail = VeilMail(api_key=os.environ["VEILMAIL_API_KEY"])3. Send Your First Email
Add an endpoint that sends a transactional email:
app/main.py
from fastapi import FastAPI
from pydantic import BaseModel
from app.veilmail_client import veilmail
app = FastAPI()
class WelcomeRequest(BaseModel):
email: str
name: str
@app.post("/api/send-welcome")
async def send_welcome(body: WelcomeRequest):
result = veilmail.emails.send(
from_address="hello@yourdomain.com",
to=body.email,
subject=f"Welcome, {body.name}!",
html=f"<h1>Welcome</h1><p>Hi {body.name}, thanks for joining.</p>",
)
return {"emailId": result["id"]}4. Send with a Template
Use templates to manage email content outside your code:
app/routes/billing.py
from datetime import datetime, timedelta
class InvoiceRequest(BaseModel):
email: str
invoice_id: str
amount: float
@app.post("/api/send-invoice")
async def send_invoice(body: InvoiceRequest):
due_date = (datetime.now() + timedelta(days=30)).strftime("%m/%d/%Y")
result = veilmail.emails.send(
from_address="billing@yourdomain.com",
to=body.email,
subject=f"Invoice #{body.invoice_id}",
template_id="template_xxxxx",
template_data={
"invoiceId": body.invoice_id,
"amount": f"${body.amount:.2f}",
"dueDate": due_date,
},
)
return {"emailId": result["id"]}5. Handle Webhooks
Receive real-time event notifications. Use await request.body() to get the raw body for signature verification.
app/webhooks.py
import hashlib
import hmac
import os
from fastapi import FastAPI, Request, HTTPException
app = FastAPI()
WEBHOOK_SECRET = os.environ["VEILMAIL_WEBHOOK_SECRET"]
def verify_signature(payload: bytes, signature: str, secret: str) -> bool:
expected = hmac.new(
secret.encode(), payload, hashlib.sha256
).hexdigest()
return hmac.compare_digest(signature, expected)
@app.post("/webhooks/veilmail")
async def handle_webhook(request: Request):
signature = request.headers.get("x-veilmail-signature", "")
body = await request.body()
if not signature or not verify_signature(body, signature, WEBHOOK_SECRET):
raise HTTPException(status_code=401, detail="Invalid signature")
event = await request.json()
match event["type"]:
case "email.delivered":
print(f"Delivered: {event['data']['emailId']}")
case "email.bounced":
print(f"Bounced: {event['data']['emailId']}")
# Remove from mailing list or flag user
case "email.complained":
print(f"Complaint: {event['data']['emailId']}")
# Unsubscribe the user
case "subscriber.unsubscribed":
print(f"Unsubscribed: {event['data']['subscriberId']}")
return {"status": "ok"}Important: Call await request.body() before await request.json() to capture the raw bytes for signature verification. FastAPI will cache the body so both calls work.
6. Manage Subscribers
Add and manage subscribers from your API:
app/routes/subscribers.py
class SubscribeRequest(BaseModel):
email: str
first_name: str
last_name: str
# Add a subscriber when a user signs up
@app.post("/api/subscribe")
async def subscribe(body: SubscribeRequest):
subscriber = veilmail.audiences.subscribers("audience_xxxxx").add(
email=body.email,
first_name=body.first_name,
last_name=body.last_name,
metadata={"source": "website"},
)
return {"subscriberId": subscriber["id"]}
class UnsubscribeRequest(BaseModel):
subscriber_id: str
# Unsubscribe
@app.post("/api/unsubscribe")
async def unsubscribe(body: UnsubscribeRequest):
veilmail.audiences.subscribers("audience_xxxxx").remove(
body.subscriber_id
)
return {"success": True}7. Error Handling
The SDK raises typed exceptions you can catch and handle:
app/routes/email.py
from veilmail import VeilMail, VeilMailError, RateLimitError
from fastapi import HTTPException
@app.post("/api/send")
async def send_email(request: Request):
body = await request.json()
try:
result = veilmail.emails.send(
from_address="hello@yourdomain.com",
to=body["email"],
subject="Hello",
html="<p>Hi!</p>",
)
return {"id": result["id"]}
except RateLimitError:
raise HTTPException(status_code=429, detail="Rate limited, try again later")
except VeilMailError as e:
raise HTTPException(status_code=e.status_code, detail=str(e))
except Exception:
raise HTTPException(status_code=500, detail="Internal error")