Sending transactional email from FastAPI

FastAPI's whole design leans on async, so the natural fit is an async HTTP client (httpx, which most FastAPI projects already pull in) rather than a blocking call inside an async request handler.

The async call

A complete send function using httpx:

• import httpx
• async with httpx.AsyncClient() as client: response = await client.post('https://api.envello.dev/emails', headers={'Authorization': f'Bearer {settings.envello_api_key}'}, json={'from': 'Acme <[email protected]>', 'to': user.email, 'subject': 'Welcome!', 'html': rendered_html})
• response.raise_for_status() to surface a failed send as an exception rather than a silently ignored error

Configuration with Pydantic Settings

Use Pydantic's BaseSettings to load and validate config from environment variables.

• class Settings(BaseSettings): envello_api_key: str, envello_from: str = 'YourApp <[email protected]>', envello_base_url: str = 'https://api.envello.dev'
• model_config = SettingsConfigDict(env_file='.env')
• Create a singleton: settings = Settings()
• Access with settings.envello_api_key throughout the app
• Pydantic validates on startup: if ENVELLO_API_KEY is missing, the app fails to start with a clear error rather than crashing on the first send attempt

Creating a reusable email service

Wrap the send logic in a class or module for dependency injection.

• Create an EmailService class that takes Settings in its constructor
• Use a shared httpx.AsyncClient (create once, reuse across requests) for connection pooling
• Define a Pydantic model for the email payload: class SendEmailRequest(BaseModel): to: EmailStr, subject: str, html: str, text: str | None = None
• The send method validates the payload with Pydantic before making the API call
• Use FastAPI's Depends() to inject the service into route handlers

Error handling

Map API errors to appropriate HTTP responses in your FastAPI endpoints.

• 400 from Envello: raise HTTPException(status_code=400, detail='Invalid email address') in your handler
• 429: raise HTTPException(status_code=429, detail='Email rate limit hit, try again later')
• 5xx: retry once with httpx's built-in retry transport, or catch and raise HTTPException(status_code=502, detail='Email service temporarily unavailable')
• httpx.ConnectError or httpx.TimeoutException: catch and raise 502
• Don't expose raw Envello error details to your API consumers; map them to your own error schema

Background sending with BackgroundTasks

For anything triggered by an incoming request (signup, password reset), use FastAPI's BackgroundTasks so the send happens after the response is already returned to the client, rather than making the user wait on Envello's API round-trip before they get a response from your endpoint.

• @router.post('/reset-password') async def reset_password(email: EmailStr, background_tasks: BackgroundTasks):
• Generate the reset token, render the email HTML
• background_tasks.add_task(email_service.send, SendEmailRequest(to=email, subject='Reset your password', html=html))
• return {'message': 'If that email exists, a reset link has been sent'}
• BackgroundTasks run after the response is sent but in the same process; they don't survive a process restart

For guaranteed delivery: use a task queue

If you need retries, persistence, and guaranteed delivery (the email must go out even if the process restarts), use a task queue like Celery, arq, or Dramatiq instead of BackgroundTasks.

• arq integrates well with FastAPI's async model: pip install arq, define a worker function, enqueue with await arq_redis.enqueue_job('send_email', to=email, subject=subject, html=html)
• arq uses Redis for persistence and handles retries with configurable backoff
• For Celery: define a task with @celery_app.task, call .delay() from the handler. Celery is synchronous by default, which is fine for I/O-bound email sends in the worker process

Webhook handling

Create a FastAPI endpoint to receive Envello delivery event webhooks.

• @router.post('/webhooks/envello') async def envello_webhook(request: Request):
• raw_body = await request.body()
• signature = request.headers.get('x-envello-signature')
• Verify HMAC-SHA256: import hmac; computed = hmac.new(secret.encode(), raw_body, 'sha256').hexdigest()
• Use hmac.compare_digest(computed, signature) to prevent timing attacks
• Parse the JSON payload and handle: delivered (log), bounced (suppress address in your DB), complained (suppress)
• Return {'received': True} with a 200 status

If you want an SDK instead of httpx directly

A Python SDK exists (pip install envello, source at github.com/Aktai-ltd/envello-python). For a FastAPI project already comfortable with httpx and Pydantic for request validation, the direct approach above is often simpler than adding another client library for a handful of send calls.

Testing with pytest

Mock the httpx client in tests to verify request payloads without making real API calls.

• Use httpx's MockTransport: transport = httpx.MockTransport(lambda req: httpx.Response(202, json={'id': 'msg_123'}))
• Inject the mock client into your EmailService during tests
• Verify the request: assert the URL, headers, and JSON body match expectations
• Test error paths: return 400, 429, and 500 responses from the mock and verify your service handles them correctly
• For integration tests, use Envello's test mode with a real API key