Come ho costruito un sistema email affidabile per FastAPI

Sistema email completo per app FastAPI, da template HTML a deliverability in produzione

---

⏱️ 12 minuti di lettura | 💻 Codice completo su GitHub → | 🎯 Livello: Intermedio

---

TL;DR

Cosa imparerai: Sistema email completo per app FastAPI, da template HTML a deliverability in produzione

Stack:

• ✅ SMTP con SendGrid/AWS SES (setup in 2 minuti)
• ✅ Template HTML responsive (Jinja2)
• ✅ Background tasks (non blocca request)
• ✅ Retry automatico
• ✅ SPF/DKIM per evitare spam folder

Quando ti serve:

• Welcome email dopo registrazione
• Password reset
• Notifiche transazionali
• Digest settimanali

Cosa NON copre:

• Marketing campaigns massive (usa Mailchimp)
• SMS/Push notifications
• Email parsing/receiving

Quick Start in 5 minuti | Template Examples | Anti-Spam Setup

---

Il Problema: Email in un’App Async

Integrare email in un’app FastAPI tocca tre problemi distinti che non sono ovvi all’inizio.

Il primo approccio naif:

@router.post("/register")
async def register(user_data: UserCreate):
    user = await user_repo.create(user_data)

    # TODO: Send welcome email
    print(f"Should send email to {user.email}")

    return user

Poi si implementa davvero, con SMTP sincrono:

@router.post("/register")
async def register(user_data: UserCreate):
    user = await user_repo.create(user_data)

    msg = MIMEText("Welcome!")
    msg['To'] = user.email

    smtp = smtplib.SMTP('smtp.gmail.com', 587)
    smtp.login('me@gmail.com', 'mypassword')  # hardcoded
    smtp.send_message(msg)  # blocca per 2-3 secondi
    smtp.quit()

    return user

I tre problemi strutturali:

1. Latenza: ogni request aspetta che l’email venga consegnata. SMTP sincrono in un handler async blocca l’event loop.
2. Deliverability: email da IP senza SPF/DKIM finiscono in spam. Gmail e Outlook applicano filtri aggressivi.
3. Resilienza: se il server SMTP è down, la request fallisce con 500. Nessun retry, email persa.

L’architettura corretta separa questi tre aspetti:

• invio asincrono via background tasks — zero blocking
• provider dedicato (SendGrid, SES) con SPF/DKIM — deliverability garantita
• retry automatico con exponential backoff — resilienza ai failure temporanei

---

Anatomia Sistema Email Moderno

┌─────────────────────────────────────┐
│   FastAPI Endpoint                  │
│   - Registrazione                   │
│   - Password reset                  │
└──────────┬──────────────────────────┘
           │
           │ Trigger background task
           ↓
┌─────────────────────────────────────┐
│   EmailService                      │
│   - Render template (Jinja2)        │
│   - Queue non-blocking send         │
└──────────┬──────────────────────────┘
           │
           │ Send async
           ↓
┌─────────────────────────────────────┐
│   SMTP Provider                     │
│   - SendGrid / AWS SES              │
│   - SPF/DKIM validation             │
└──────────┬──────────────────────────┘
           │
           ↓
       [User Inbox] 📧

Componenti chiave:

1. SMTP Provider - SendGrid/SES, non Gmail
2. Jinja2 Templates - HTML professionale
3. Background Tasks - FastAPI native, zero blocking
4. Retry Logic - 3 tentativi con backoff
5. DNS Records - SPF/DKIM per deliverability

---

Quick Start (5 Minuti)

Step 1: SMTP Provider (2 minuti)

Raccomandazione: SendGrid (100 email/giorno free)

# 1. Signup su sendgrid.com
# 2. Create API Key (Settings → API Keys)
# 3. Aggiungi a .env

# .env
SMTP_HOST=smtp.sendgrid.net
SMTP_PORT=587
SMTP_USER=apikey
SMTP_PASSWORD=SG.your_actual_api_key_here
SMTP_FROM_EMAIL=noreply@yourdomain.com
SMTP_FROM_NAME="Your App"

Alternative:

• AWS SES - economico a volume (€0.10 per 1000 email)
• Mailgun - ottimo per transazionali

Step 2: Install Dependencies (30 secondi)

pip install fastapi aiosmtplib jinja2 pydantic-settings beautifulsoup4

Step 3: EmailService Base (2 minuti)

# app/services/email_service.py

import aiosmtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from pathlib import Path
from jinja2 import Environment, FileSystemLoader
from app.config import settings

class EmailService:
    def __init__(self):
        # Setup Jinja2 templates
        template_dir = Path(__file__).parent.parent / "templates" / "emails"
        self.jinja_env = Environment(
            loader=FileSystemLoader(str(template_dir)),
            autoescape=True
        )

    async def send_email(
        self,
        to_email: str,
        subject: str,
        html_content: str
    ):
        """Send HTML email via SMTP."""

        message = MIMEMultipart("alternative")
        message["Subject"] = subject
        message["From"] = f"{settings.SMTP_FROM_NAME} <{settings.SMTP_FROM_EMAIL}>"
        message["To"] = to_email

        # Attach HTML
        html_part = MIMEText(html_content, "html")
        message.attach(html_part)

        # Send async
        await aiosmtplib.send(
            message,
            hostname=settings.SMTP_HOST,
            port=settings.SMTP_PORT,
            username=settings.SMTP_USER,
            password=settings.SMTP_PASSWORD,
            start_tls=True
        )

Step 4: Use in Endpoint (30 secondi)

# app/routers/auth.py

from fastapi import BackgroundTasks
from app.services.email_service import EmailService

email_service = EmailService()

@router.post("/register")
async def register(
    user_data: UserCreate,
    background_tasks: BackgroundTasks
):
    # Create user
    user = await user_repo.create(user_data)

    # Send email in background (non-blocking!)
    background_tasks.add_task(
        email_service.send_welcome_email,
        user
    )

    return user  # Risposta immediata!

Done! Request ritorna subito, email inviata in background.

---

Template HTML con Jinja2

Base Template Responsive

<!-- templates/emails/base.html -->
<!DOCTYPE html>
<html>
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <style>
        body {
            margin: 0;
            padding: 0;
            font-family: Arial, sans-serif;
            background-color: #f4f4f4;
        }
        .email-container {
            max-width: 600px;
            margin: 20px auto;
            background: #ffffff;
            border-radius: 8px;
            overflow: hidden;
        }
        .header {
            background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
            padding: 30px 20px;
            text-align: center;
            color: #ffffff;
        }
        .header h1 {
            margin: 0;
            font-size: 24px;
        }
        .content {
            padding: 30px 25px;
            color: #333333;
            line-height: 1.6;
        }
        .button {
            display: inline-block;
            padding: 12px 24px;
            margin: 15px 0;
            background-color: #667eea;
            color: #ffffff !important;
            text-decoration: none;
            border-radius: 5px;
            font-weight: bold;
        }
        .footer {
            background-color: #f8f9fa;
            padding: 20px;
            text-align: center;
            font-size: 12px;
            color: #6c757d;
        }
        @media only screen and (max-width: 600px) {
            .content { padding: 20px 15px !important; }
            .header h1 { font-size: 20px !important; }
        }
    </style>
</head>
<body>
    <div class="email-container">
        <div class="header">
            {% block header %}
            <h1>Your App</h1>
            {% endblock %}
        </div>

        <div class="content">
            {% block content %}{% endblock %}
        </div>

        <div class="footer">
            {% block footer %}
            <p>© 2025 Your App. Tutti i diritti riservati.</p>
            <p><a href="{{ unsubscribe_url }}">Cancellati</a></p>
            {% endblock %}
        </div>
    </div>
</body>
</html>

Welcome Email

<!-- templates/emails/welcome.html -->
{% extends "base.html" %}

{% block header %}
<h1>👋 Benvenuto, {{ username }}!</h1>
{% endblock %}

{% block content %}
<h2>Il tuo account è pronto!</h2>

<p>Ciao <strong>{{ username }}</strong>,</p>

<p>
    Grazie per esserti registrato! Il tuo account è stato creato con successo.
</p>

<p>
    Per iniziare, conferma il tuo indirizzo email:
</p>

<center>
    <a href="{{ confirm_url }}" class="button">
        Conferma Email
    </a>
</center>

<p style="font-size: 14px; color: #6c757d;">
    Se non hai richiesto questa registrazione, ignora questa email.
</p>

<p>
    Buon divertimento! 🎮
</p>
{% endblock %}

Password Reset

<!-- templates/emails/password_reset.html -->
{% extends "base.html" %}

{% block header %}
<h1>🔐 Reset Password</h1>
{% endblock %}

{% block content %}
<h2>Richiesta reset password</h2>

<p>Ciao <strong>{{ username }}</strong>,</p>

<p>
    Abbiamo ricevuto una richiesta per resettare la password del tuo account.
</p>

<center>
    <a href="{{ reset_url }}" class="button">
        Reset Password
    </a>
</center>

<div style="background: #fff3cd; border-left: 4px solid #ffc107; padding: 15px; margin: 20px 0;">
    <strong>⚠️ Importante:</strong> Questo link è valido per <strong>1 ora</strong>.
</div>

<p>
    Se non hai richiesto tu questa operazione, ignora questa email.
</p>
{% endblock %}

EmailService con Templates

# app/services/email_service.py

class EmailService:
    def render_template(self, template_name: str, **context) -> str:
        """Render Jinja2 template."""
        template = self.jinja_env.get_template(template_name)
        return template.render(**context)

    async def send_welcome_email(self, user: User):
        """Send welcome email con conferma account."""

        # Generate confirm token
        confirm_token = generate_token(user.id)

        # Render template
        html_content = self.render_template(
            "welcome.html",
            username=user.username,
            confirm_url=f"{settings.APP_URL}/confirm/{confirm_token}",
            unsubscribe_url=f"{settings.APP_URL}/unsubscribe/{user.id}"
        )

        # Send
        await self.send_email(
            to_email=user.email,
            subject=f"Benvenuto, {user.username}! 👋",
            html_content=html_content
        )

    async def send_password_reset(self, user: User, reset_token: str):
        """Send password reset email."""

        html_content = self.render_template(
            "password_reset.html",
            username=user.username,
            reset_url=f"{settings.APP_URL}/reset-password/{reset_token}",
            unsubscribe_url=f"{settings.APP_URL}/unsubscribe/{user.id}"
        )

        await self.send_email(
            to_email=user.email,
            subject="🔐 Reset Password Richiesto",
            html_content=html_content
        )

---

Background Tasks & Retry

Problema: Email Blocking

# BAD: Request waits for email (3 seconds!)
@router.post("/register")
async def register(user_data: UserCreate):
    user = await user_repo.create(user_data)
    await email_service.send_welcome_email(user)  # BLOCCA!
    return user

# GOOD: Background, risposta immediata
@router.post("/register")
async def register(user_data: UserCreate, background_tasks: BackgroundTasks):
    user = await user_repo.create(user_data)
    background_tasks.add_task(email_service.send_welcome_email, user)
    return user  # Risponde subito!

Retry Logic con Exponential Backoff

# app/services/email_service.py

import asyncio
import logging

logger = logging.getLogger(__name__)

class EmailService:
    async def send_email_with_retry(
        self,
        to_email: str,
        subject: str,
        html_content: str,
        max_retries: int = 3
    ) -> bool:
        """Send email con retry automatico."""

        for attempt in range(max_retries):
            try:
                await self.send_email(to_email, subject, html_content)

                logger.info(
                    f"✅ Email sent to {to_email}",
                    extra={"attempt": attempt + 1}
                )
                return True

            except aiosmtplib.SMTPException as e:
                logger.warning(
                    f"⚠️ Email failed (attempt {attempt + 1}/{max_retries}): {e}",
                    extra={"to": to_email}
                )

                # Last attempt failed
                if attempt == max_retries - 1:
                    logger.error(f"❌ Email failed after {max_retries} attempts")
                    return False

                # Exponential backoff: 1s, 2s, 4s
                wait_time = 2 ** attempt
                await asyncio.sleep(wait_time)

        return False

    # Use retry for all send methods
    async def send_welcome_email(self, user: User):
        html_content = self.render_template("welcome.html", ...)

        success = await self.send_email_with_retry(
            to_email=user.email,
            subject="Benvenuto!",
            html_content=html_content
        )

        if not success:
            # Log for monitoring
            logger.error(f"Failed to send welcome email to {user.email}")

Perché funziona:

• 1° tentativo fallisce → aspetta 1s → riprova
• 2° tentativo fallisce → aspetta 2s → riprova
• 3° tentativo fallisce → log error, stop

Risolve 90% dei failure temporanei (SMTP timeout, network glitch).

---

Deliverability: Evitare lo Spam

Il Problema

Le tue email finiscono nello spam perché:

1. ❌ Dominio non verificato
2. ❌ Niente SPF/DKIM
3. ❌ IP reputation bassa
4. ❌ Contenuto “spammy”

SPF/DKIM Setup (15 minuti)

SPF (Sender Policy Framework)

Dice a Gmail/Outlook: “Solo SendGrid può inviare email per mio conto”

# Add TXT DNS record for your domain
Name: @
Type: TXT
Value: v=spf1 include:sendgrid.net ~all
TTL: 3600

DKIM (DomainKeys Identified Mail)

Firma digitale che verifica autenticità email.

# SendGrid provides these CNAME records
Name: s1._domainkey.yourdomain.com
Type: CNAME
Value: s1.domainkey.u123456.wl.sendgrid.net

Name: s2._domainkey.yourdomain.com
Type: CNAME
Value: s2.domainkey.u123456.wl.sendgrid.net

Verifica Setup:

1. Vai su mxtoolbox.com/SuperTool.aspx
2. Inserisci yourdomain.com
3. Check SPF/DKIM records

Risultato: Da 60% delivery → 98%+ delivery.

Best Practices Anti-Spam

# app/services/email_service.py

class EmailService:
    # 1. Sempre unsubscribe link (passare user_id esplicitamente)
    async def send_email(self, to_email: str, subject: str, html: str, user_id: str = ""):
        # Include in footer
        unsubscribe = f"{settings.APP_URL}/unsubscribe/{user_id}"

    # 2. Plain text alternative (migliora deliverability)
    def create_message(self, html_content: str):
        message = MIMEMultipart("alternative")

        # Plain text version
        text_content = self.html_to_text(html_content)
        message.attach(MIMEText(text_content, "plain"))

        # HTML version
        message.attach(MIMEText(html_content, "html"))

        return message

    def html_to_text(self, html: str) -> str:
        """Convert HTML to plain text (basic)."""
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(html, "html.parser")
        return soup.get_text()

Checklist:

• ✅ SPF/DKIM records configurati
• ✅ Unsubscribe link in ogni email
• ✅ Physical address in footer (GDPR)
• ✅ Plain text alternative
• ✅ Evita parole spam (FREE!!! CLICK NOW!!!)
• ✅ Ratio testo/immagini > 60% testo
• ✅ Non usare URL shortener

---

Testing

Mock per Development

# app/services/email_service.py

class EmailService:
    def __init__(self):
        self.mock = settings.ENVIRONMENT == "development"

    async def send_email(self, to_email: str, subject: str, html: str):
        if self.mock:
            # Save to file instead of sending
            self._save_to_file(to_email, subject, html)
            logger.info(f"📧 [MOCK] Email saved to file: {to_email}")
            return

        # Real send in production
        await aiosmtplib.send(...)

    def _save_to_file(self, to_email: str, subject: str, html: str):
        """Save email as HTML file for review."""
        import os
        from datetime import datetime

        os.makedirs("emails_debug", exist_ok=True)

        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"emails_debug/{timestamp}_{to_email.replace('@', '_')}.html"

        with open(filename, "w") as f:
            f.write(f"<!-- To: {to_email} -->\n")
            f.write(f"<!-- Subject: {subject} -->\n\n")
            f.write(html)

Apri emails_debug/ e vedi le email generate. Perfetto per development.

Unit Test Base

# tests/test_email_service.py

import pytest
from unittest.mock import patch, AsyncMock
from app.services.email_service import EmailService

@pytest.fixture
def email_service():
    return EmailService()

@pytest.mark.asyncio
async def test_render_welcome_template(email_service):
    """Test template rendering."""

    html = email_service.render_template(
        "welcome.html",
        username="TestUser",
        confirm_url="https://example.com/confirm/token"
    )

    assert "Benvenuto, TestUser!" in html
    assert "confirm/token" in html

@pytest.mark.asyncio
@patch('aiosmtplib.send')
async def test_send_email(mock_send, email_service):
    """Test email sending (mocked)."""

    mock_send.return_value = None

    await email_service.send_email(
        to_email="test@example.com",
        subject="Test",
        html_content="<h1>Test</h1>"
    )

    # Verify SMTP called
    mock_send.assert_called_once()

---

Errori Comuni (Che Ho Fatto)

1. Password Hardcoded

# NEVER
SMTP_PASSWORD = "mypassword123"

# ALWAYS
SMTP_PASSWORD = os.getenv("SMTP_PASSWORD")

Mi è costato: Account compromesso, spam inviate dal mio IP.

2. Niente Rate Limiting

# No rate limiting
@router.post("/send-email")
async def send_email(email: str):
    await email_service.send(email)

Risultato: 10,000 email in 1 minuto → account bannato.

✅ Fix:

from slowapi import Limiter

limiter = Limiter(key_func=get_remote_address)

@router.post("/send-email")
@limiter.limit("5/minute")
async def send_email(request: Request, email: str):
    await email_service.send(email)

3. Dimenticare Unsubscribe

# GDPR violation, ends up in spam
# No way to unsubscribe

# Required in every email
<a href="{{ unsubscribe_url }}">Cancellati</a>

4. SMTP Sincrono

# Blocks request
await email_service.send(email)  # 3 seconds!

# Background
background_tasks.add_task(email_service.send, email)

5. Niente Logging

# If it fails, you won't know
await email_service.send(email)

# Always log
logger.info(f"Sending email to {email}")
try:
    await email_service.send(email)
    logger.info(f"✅ Email sent to {email}")
except Exception as e:
    logger.error(f"❌ Email failed: {email}, error: {e}")

---

Conclusione

Next Steps

Quick Start:

git clone https://github.com/layerbylayer-blog/fastapi-email-system
cd fastapi-email-system

# Copia .env.example e configura
cp .env.example .env

# Run
uvicorn app.main:app --reload

Nel Tuo Progetto:

Week 1:

1. Setup SendGrid account
2. Configura SPF/DKIM DNS records
3. Implementa EmailService base
4. Crea 1-2 template essenziali

Week 2: 5. Background tasks tutti gli endpoint 6. Retry logic 7. Testing mock 8. Deploy e monitor

---

Risorse

GitHub: fastapi-email-system

Include:

• ✅ EmailService completo
• ✅ 5+ template pronti
• ✅ Testing suite
• ✅ Docker compose
• ✅ Environment setup

Tools:

• SendGrid - SMTP provider (100 email/day free)
• AWS SES - Production scalable
• Mail Tester - Test spam score
• MXToolbox - Verifica DNS/SPF/DKIM

Prossimo articolo: Logging strutturato con context e correlation IDs