FastAPI

This guide shows you how to integrate Unsent into your FastAPI application for high-performance, asynchronous email sending.

Prerequisites

Python 3.8+: FastAPI requires Python 3.8 or higher

FastAPI 0.68+: This guide assumes modern FastAPI with async support

Unsent API Key: Generate one in your Unsent dashboard

Verified Domain: Set up a domain in the Domains section

Installation

Install FastAPI, Unsent SDK, and ASGI server:

pip install fastapi unsent uvicorn python-dotenv pydantic-settings

Configuration

Environment Variables

Create a .env file:

.env

UNSENT_API_KEY=un_...
UNSENT_FROM_EMAIL=noreply@yourdomain.com
APP_NAME=My FastAPI App
APP_URL=http://localhost:8000

Settings with Pydantic

config.py

from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    unsent_api_key: str
    unsent_from_email: str
    app_name: str = "FastAPI App"
    app_url: str = "http://localhost:8000"
    
    class Config:
        env_file = ".env"

@lru_cache()
def get_settings():
    return Settings()

Quick Start

Basic FastAPI App

main.py

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, EmailStr
from unsent import unsent, types
from config import get_settings, Settings

app = FastAPI(title="Email Service API")

# Initialize Unsent client
def get_unsent_client(settings: Settings = Depends(get_settings)):
    return unsent(settings.unsent_api_key, raise_on_error=False)

class EmailRequest(BaseModel):
    to: EmailStr
    subject: str
    message: str

@app.post("/send-email")
async def send_email(
    request: EmailRequest,
    client: unsent = Depends(get_unsent_client),
    settings: Settings = Depends(get_settings)
):
    """Send a simple email."""
    payload: types.EmailCreate = {
        "to": request.to,
        "from": settings.unsent_from_email,
        "subject": request.subject,
        "html": f"<h1>{request.message}</h1>",
        "text": request.message
    }
    
    data, error = client.emails.send(payload)
    
    if error:
        raise HTTPException(status_code=500, detail=f"Failed to send email: {error}")
    
    return {"success": True, "email_id": data.get('id')}

@app.get("/")
async def root():
    return {"message": "FastAPI + Unsent Email Service"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Run the server:

uvicorn main:app --reload

Dependency Injection

FastAPI's dependency injection makes it easy to manage the Unsent client:

dependencies.py

from unsent import unsent
from config import get_settings
from functools import lru_cache

@lru_cache()
def get_unsent_client():
    """Create a singleton Unsent client."""
    settings = get_settings()
    return unsent(settings.unsent_api_key, raise_on_error=False)

Use in routes:

main.py

from fastapi import Depends
from dependencies import get_unsent_client

@app.post("/send-email")
async def send_email(
    request: EmailRequest,
    client: unsent = Depends(get_unsent_client)
):
    # Use client...
    pass

Email Service Layer

Create a dedicated service for email operations:

services/email_service.py

from unsent import unsent, types
from typing import Dict, Optional, Tuple
from jinja2 import Environment, FileSystemLoader, select_autoescape
import os

class EmailService:
    def __init__(self, client: unsent, from_email: str):
        self.client = client
        self.from_email = from_email
        
        # Setup Jinja2 for templates
        self.jinja_env = Environment(
            loader=FileSystemLoader('templates/emails'),
            autoescape=select_autoescape(['html', 'xml'])
        )
    
    async def send_welcome_email(
        self,
        user_email: str,
        user_name: str,
        app_url: str
    ) -> Tuple[Optional[Dict], Optional[Dict]]:
        """Send welcome email using template."""
        template = self.jinja_env.get_template('welcome.html')
        html = template.render(
            user_name=user_name,
            app_url=app_url
        )
        
        payload: types.EmailCreate = {
            "to": user_email,
            "from": self.from_email,
            "subject": "Welcome to Our Platform!",
            "html": html,
            "text": f"Welcome, {user_name}! Thanks for joining us."
        }
        
        return self.client.emails.send(payload)
    
    async def send_password_reset(
        self,
        user_email: str,
        reset_token: str,
        app_url: str
    ) -> Tuple[Optional[Dict], Optional[Dict]]:
        """Send password reset email."""
        reset_url = f"{app_url}/reset-password?token={reset_token}"
        
        template = self.jinja_env.get_template('password_reset.html')
        html = template.render(
            reset_url=reset_url,
            expires_in="1 hour"
        )
        
        payload: types.EmailCreate = {
            "to": user_email,
            "from": self.from_email,
            "subject": "Reset Your Password",
            "html": html,
            "text": f"Reset your password: {reset_url}"
        }
        
        return self.client.emails.send(payload)
    
    async def send_verification_email(
        self,
        user_email: str,
        verification_code: str
    ) -> Tuple[Optional[Dict], Optional[Dict]]:
        """Send email verification code."""
        payload: types.EmailCreate = {
            "to": user_email,
            "from": self.from_email,
            "subject": "Verify Your Email",
            "html": f"""
                <h2>Email Verification</h2>
                <p>Your verification code is: <strong>{verification_code}</strong></p>
                <p>This code will expire in 10 minutes.</p>
            """,
            "text": f"Your verification code is: {verification_code}"
        }
        
        return self.client.emails.send(payload)

# Dependency to get email service
def get_email_service(
    client: unsent = Depends(get_unsent_client),
    settings: Settings = Depends(get_settings)
) -> EmailService:
    return EmailService(client, settings.unsent_from_email)

Router Organization

Structure your API with routers:

routers/auth.py

from fastapi import APIRouter, HTTPException, Depends, BackgroundTasks
from pydantic import BaseModel, EmailStr
from services.email_service import EmailService, get_email_service
from config import get_settings, Settings
import secrets

router = APIRouter(prefix="/auth", tags=["Authentication"])

class RegisterRequest(BaseModel):
    email: EmailStr
    name: str
    password: str

class ForgotPasswordRequest(BaseModel):
    email: EmailStr

@router.post("/register")
async def register(
    request: RegisterRequest,
    background_tasks: BackgroundTasks,
    email_service: EmailService = Depends(get_email_service),
    settings: Settings = Depends(get_settings)
):
    """Register a new user and send welcome email."""
    # Create user in database (simplified)
    # user = create_user(request.email, request.name, request.password)
    
    # Send welcome email in background
    background_tasks.add_task(
        email_service.send_welcome_email,
        request.email,
        request.name,
        settings.app_url
    )
    
    return {
        "message": "User registered successfully",
        "email": request.email
    }

@router.post("/forgot-password")
async def forgot_password(
    request: ForgotPasswordRequest,
    background_tasks: BackgroundTasks,
    email_service: EmailService = Depends(get_email_service),
    settings: Settings = Depends(get_settings)
):
    """Send password reset email."""
    # Generate reset token
    reset_token = secrets.token_urlsafe(32)
    
    # Store token in database/cache
    # await store_reset_token(request.email, reset_token)
    
    # Send reset email in background
    background_tasks.add_task(
        email_service.send_password_reset,
        request.email,
        reset_token,
        settings.app_url
    )
    
    return {"message": "Password reset email sent"}

@router.post("/verify-email")
async def send_verification(
    email: EmailStr,
    background_tasks: BackgroundTasks,
    email_service: EmailService = Depends(get_email_service)
):
    """Send email verification code."""
    verification_code = ''.join([str(secrets.randbelow(10)) for _ in range(6)])
    
    # Store code in cache
    # await cache.set(f"verify:{email}", verification_code, ex=600)
    
    # Send verification email
    background_tasks.add_task(
        email_service.send_verification_email,
        email,
        verification_code
    )
    
    return {"message": "Verification code sent"}

main.py

from routers import auth

app.include_router(auth.router)

Background Tasks

Using FastAPI Background Tasks

from fastapi import BackgroundTasks

@app.post("/send-newsletter")
async def send_newsletter(
    background_tasks: BackgroundTasks,
    client: unsent = Depends(get_unsent_client)
):
    """Send newsletter to all subscribers."""
    # Get subscribers from database
    subscribers = ["user1@example.com", "user2@example.com"]
    
    # Queue batch email in background
    background_tasks.add_task(send_batch_emails, client, subscribers)
    
    return {"message": "Newsletter queued"}

async def send_batch_emails(client: unsent, subscribers: list):
    """Background task to send batch emails."""
    batch_items = [
        {
            "to": email,
            "from": os.getenv("UNSENT_FROM_EMAIL"),
            "subject": "Newsletter",
            "html": "<h1>Latest News</h1>"
        }
        for email in subscribers
    ]
    
    data, error = client.emails.batch(batch_items)
    if error:
        print(f"Batch send failed: {error}")

Using Celery with FastAPI

celery_worker.py

from celery import Celery
from unsent import unsent, types
import os

celery_app = Celery(
    'tasks',
    broker=os.getenv('REDIS_URL', 'redis://localhost:6379/0'),
    backend=os.getenv('REDIS_URL', 'redis://localhost:6379/0')
)

client = unsent(os.getenv('UNSENT_API_KEY'), raise_on_error=False)

@celery_app.task(bind=True, max_retries=3)
def send_email_task(self, email_data: dict):
    """Celery task to send email."""
    try:
        data, error = client.emails.send(email_data)
        if error:
            raise Exception(f"Failed to send email: {error}")
        return data
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60 * (2 ** self.request.retries))

Use in FastAPI:

main.py

from celery_worker import send_email_task

@app.post("/send-async")
async def send_async_email(request: EmailRequest):
    """Send email via Celery."""
    email_data = {
        "to": request.to,
        "from": os.getenv("UNSENT_FROM_EMAIL"),
        "subject": request.subject,
        "html": f"<h1>{request.message}</h1>"
    }
    
    task = send_email_task.delay(email_data)
    return {"task_id": task.id, "status": "queued"}

Webhooks

Handle Unsent webhooks with proper validation:

routers/webhooks.py

from fastapi import APIRouter, Request, HTTPException
import logging

router = APIRouter(prefix="/webhooks", tags=["Webhooks"])
logger = logging.getLogger(__name__)

@router.post("/unsent")
async def handle_unsent_webhook(request: Request):
    """Handle incoming Unsent webhook events."""
    try:
        event = await request.json()
        event_type = event.get('type')
        data = event.get('data', {})
        
        # Route to appropriate handler
        handlers = {
            'email.delivered': handle_email_delivered,
            'email.bounced': handle_email_bounced,
            'email.opened': handle_email_opened,
            'email.clicked': handle_email_clicked,
            'email.complained': handle_email_complained
        }
        
        handler = handlers.get(event_type)
        if handler:
            await handler(data)
        else:
            logger.info(f"Unhandled webhook event: {event_type}")
        
        return {"status": "received"}
    
    except Exception as e:
        logger.error(f"Webhook processing error: {e}")
        raise HTTPException(status_code=500, detail=str(e))

async def handle_email_delivered(data: dict):
    """Handle email delivered event."""
    email_id = data.get('emailId')
    recipient = data.get('recipient')
    logger.info(f"Email {email_id} delivered to {recipient}")
    # Update database, track analytics

async def handle_email_bounced(data: dict):
    """Handle email bounced event."""
    email_id = data.get('emailId')
    bounce_type = data.get('bounceType')
    logger.warning(f"Email {email_id} bounced: {bounce_type}")
    # Mark email as bounced, update suppression list

async def handle_email_opened(data: dict):
    """Handle email opened event."""
    email_id = data.get('emailId')
    logger.info(f"Email {email_id} opened")
    # Track open rate

async def handle_email_clicked(data: dict):
    """Handle link clicked event."""
    email_id = data.get('emailId')
    link = data.get('link')
    logger.info(f"Link clicked in {email_id}: {link}")
    # Track click rate

async def handle_email_complained(data: dict):
    """Handle spam complaint."""
    recipient = data.get('recipient')
    logger.warning(f"Spam complaint from {recipient}")
    # Unsubscribe user automatically

Testing

Pytest with FastAPI TestClient

tests/test_email_endpoints.py

import pytest
from fastapi.testclient import TestClient
from unittest.mock import Mock, patch
from main import app

client = TestClient(app)

@pytest.fixture
def mock_unsent():
    with patch('dependencies.unsent') as mock:
        mock_client = Mock()
        mock.return_value = mock_client
        yield mock_client

def test_send_email_success(mock_unsent):
    """Test successful email send."""
    mock_unsent.emails.send.return_value = ({"id": "email_123"}, None)
    
    response = client.post(
        "/send-email",
        json={
            "to": "test@example.com",
            "subject": "Test",
            "message": "Hello"
        }
    )
    
    assert response.status_code == 200
    assert response.json()["success"] is True
    assert response.json()["email_id"] == "email_123"

def test_send_email_failure(mock_unsent):
    """Test failed email send."""
    mock_unsent.emails.send.return_value = (None, {"message": "API Error"})
    
    response = client.post(
        "/send-email",
        json={
            "to": "test@example.com",
            "subject": "Test",
            "message": "Hello"
        }
    )
    
    assert response.status_code == 500
    assert "Failed to send email" in response.json()["detail"]

def test_register_with_email(mock_unsent):
    """Test user registration with welcome email."""
    mock_unsent.emails.send.return_value = ({"id": "email_456"}, None)
    
    response = client.post(
        "/auth/register",
        json={
            "email": "newuser@example.com",
            "name": "New User",
            "password": "securepass123"
        }
    )
    
    assert response.status_code == 200
    assert "registered successfully" in response.json()["message"]

@pytest.mark.asyncio
async def test_email_service():
    """Test email service directly."""
    from services.email_service import EmailService
    
    mock_client = Mock()
    mock_client.emails.send.return_value = ({"id": "test_123"}, None)
    
    service = EmailService(mock_client, "from@example.com")
    result, error = await service.send_welcome_email(
        "user@example.com",
        "Test User",
        "http://example.com"
    )
    
    assert error is None
    assert result["id"] == "test_123"

Run tests:

pytest tests/ -v

Best Practices

1. Proper Error Handling

from fastapi import HTTPException
from unsent import unsentHTTPError

@app.post("/send-email")
async def send_email(request: EmailRequest):
    try:
        client = get_unsent_client()
        data, error = client.emails.send(payload)
        
        if error:
            # Log error details
            logger.error(f"Email send failed: {error}")
            raise HTTPException(
                status_code=500,
                detail={"error": "Failed to send email", "details": error}
            )
        
        return {"success": True, "email_id": data['id']}
    
    except unsentHTTPError as e:
        logger.error(f"Unsent API error: {e}")
        raise HTTPException(status_code=502, detail="Email service unavailable")
    except Exception as e:
        logger.exception("Unexpected error")
        raise HTTPException(status_code=500, detail="Internal server error")

2. Request Validation with Pydantic

from pydantic import BaseModel, EmailStr, Field, validator
from typing import Optional, List

class EmailRequest(BaseModel):
    to: EmailStr | List[EmailStr]
    subject: str = Field(..., min_length=1, max_length=200)
    message: str = Field(..., min_length=1)
    reply_to: Optional[EmailStr] = None
    
    @validator('to')
    def validate_recipients(cls, v):
        if isinstance(v, list) and len(v) > 50:
            raise ValueError('Cannot send to more than 50 recipients at once')
        return v

3. Response Models

from pydantic import BaseModel

class EmailResponse(BaseModel):
    success: bool
    email_id: str
    message: str = "Email sent successfully"

@app.post("/send-email", response_model=EmailResponse)
async def send_email(request: EmailRequest):
    # ... send email logic
    return EmailResponse(
        success=True,
        email_id=data['id']
    )

4. Middleware for Logging

from fastapi import Request
import time
import logging

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start_time = time.time()
    
    response = await call_next(request)
    
    process_time = time.time() - start_time
    logger.info(
        f"{request.method} {request.url.path} "
        f"completed in {process_time:.3f}s with status {response.status_code}"
    )
    
    return response

Next Steps

Python SDK: Explore all SDK features
Flask Guide: Sync email sending with Flask
Django Guide: Full Django integration

FastAPI's async capabilities make it perfect for high-throughput email sending. Use background tasks for non-critical emails and Celery for guaranteed delivery.

FastAPI

On this page