Python smtplib in produzione: timeout, retry, idempotency

La libreria smtplib della standard library Python è uno strumento minimo ma sufficiente per inviare email da uno script. Quando però viene messa in produzione senza wrapping, emergono problemi: timeout di default illimitati che congelano il worker, SSL context che accetta certificati invalidi, retry implementati come ciclo while True senza backoff, doppi invii in caso di network glitch. In questo articolo vediamo come usare smtplib in modo production-grade, configurare SSL context restrittivi, implementare retry idempotenti con backoff esponenziale, e per workload async-heavy passare ad aiosmtplib. Tutti i pattern sono ispirati a setup reali in scaling oltre 50k invii/giorno.

Lo script minimo (che NON va in produzione)

import smtplib
from email.message import EmailMessage

msg = EmailMessage()
msg["From"] = "noreply@targetsmtp.it"
msg["To"] = "cliente@example.it"
msg["Subject"] = "Conferma ordine #4287"
msg.set_content("Grazie per il tuo ordine...")

with smtplib.SMTP("smtp.targetsmtp.it", 587) as s:
    s.starttls()
    s.login("username", "password")
    s.send_message(msg)

Funziona, ma in produzione ha tre problemi:

1. Nessun timeout: se il server SMTP non risponde, il processo si blocca indefinitamente
2. starttls() senza SSL context: accetta certificati invalidi/scaduti silenziosamente
3. Nessun retry, nessuna idempotency: failure transitorio = messaggio perso o ritentato N volte (duplicati)

Configurazione production-grade

import smtplib
import ssl
from email.message import EmailMessage

def build_ssl_context() -> ssl.SSLContext:
    ctx = ssl.create_default_context()
    ctx.minimum_version = ssl.TLSVersion.TLSv1_2
    ctx.check_hostname = True
    ctx.verify_mode = ssl.CERT_REQUIRED
    return ctx

def send_email(msg: EmailMessage, host: str, port: int = 587,
               user: str = "", password: str = "", timeout: float = 15.0):
    ctx = build_ssl_context()
    with smtplib.SMTP(host, port, timeout=timeout) as s:
        s.ehlo()
        s.starttls(context=ctx)
        s.ehlo()
        if user:
            s.login(user, password)
        s.send_message(msg)

Punti chiave

• timeout=15.0: limite per ogni operazione SMTP (connect, helo, mail from, rcpt, data). 15 secondi è il valore standard per transactional. Per bulk può salire a 30
• ssl.create_default_context(): usa il bundle di CA system, verifica hostname, rifiuta cert invalidi
• minimum_version = TLSv1_2: SSLv3 e TLSv1.0/1.1 sono insicuri, vanno bloccati
• Due ehlo(): il primo prima di STARTTLS, il secondo dopo (richiesto dall'RFC perché le extension possono cambiare post-TLS)

⚠️ Attenzione: smtplib.SMTP_SSL con port 465 NON richiede starttls(): la connessione è già TLS dal primo byte. Confondere SMTP+STARTTLS (587) con SMTPS (465) è errore classico. Per moderne infrastrutture preferisci 587+STARTTLS.

Retry con backoff esponenziale

import random
import time
from typing import Callable
import smtplib

RETRYABLE_EXCEPTIONS = (
    smtplib.SMTPServerDisconnected,
    smtplib.SMTPConnectError,
    smtplib.SMTPHeloError,
    TimeoutError,
    ConnectionError,
)

def is_retryable_smtp_error(exc: smtplib.SMTPException) -> bool:
    if isinstance(exc, smtplib.SMTPResponseException):
        return 400 <= exc.smtp_code < 500
    return isinstance(exc, RETRYABLE_EXCEPTIONS)

def with_retry(fn: Callable, max_attempts: int = 5):
    for attempt in range(max_attempts):
        try:
            return fn()
        except smtplib.SMTPException as e:
            if not is_retryable_smtp_error(e) or attempt == max_attempts - 1:
                raise
            delay = min(2 ** attempt + random.uniform(0, 1), 60)
            time.sleep(delay)

Tre regole implementate:

1. 5xx non si ritentano: la regex 400 <= code < 500 filtra solo i transitori
2. Backoff esponenziale + jitter: 2^attempt + random(0,1) previene thundering herd
3. Cap a 60 secondi: oltre, conviene mettere in queue e ritentare più tardi

Idempotency: pattern con DB tracker

Senza idempotency, il retry post-network-glitch genera duplicati. Pattern: persistere un tracker prima dell'invio, controllarlo prima di ritentare.

import hashlib
import json
import sqlite3
from contextlib import contextmanager
from email.message import EmailMessage

def idempotency_key(msg: EmailMessage, context: dict | None = None) -> str:
    payload = {
        "from": msg["From"],
        "to": msg["To"],
        "subject": msg["Subject"],
        "body_hash": hashlib.sha256(msg.get_content().encode()).hexdigest(),
        "ctx": context or {},
    }
    return hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()[:32]

@contextmanager
def idempotent_send(db: sqlite3.Connection, key: str):
    row = db.execute(
        "SELECT status FROM email_attempts WHERE key = ?", (key,)
    ).fetchone()
    if row and row[0] == "sent":
        raise RuntimeError(f"Already sent: {key}")
    db.execute(
        "INSERT OR REPLACE INTO email_attempts(key, status, ts) VALUES (?, 'pending', datetime('now'))",
        (key,),
    )
    db.commit()
    try:
        yield
        db.execute(
            "UPDATE email_attempts SET status = 'sent', ts = datetime('now') WHERE key = ?",
            (key,),
        )
        db.commit()
    except Exception:
        db.execute(
            "UPDATE email_attempts SET status = 'failed' WHERE key = ?", (key,)
        )
        db.commit()
        raise

Async: aiosmtplib

Per workload async-heavy (FastAPI, microservizi), bloccare un event loop in attesa di SMTP non è accettabile. aiosmtplib espone API asyncio compatibili:

import asyncio
import ssl
from email.message import EmailMessage
import aiosmtplib

async def send_async(msg: EmailMessage):
    ctx = ssl.create_default_context()
    ctx.minimum_version = ssl.TLSVersion.TLSv1_2
    await aiosmtplib.send(
        msg,
        hostname="smtp.targetsmtp.it",
        port=587,
        start_tls=True,
        tls_context=ctx,
        username="user",
        password="pass",
        timeout=15,
    )

# Invio batch parallelo controllato
async def send_batch(messages: list[EmailMessage], concurrency: int = 10):
    sem = asyncio.Semaphore(concurrency)
    async def _send(m):
        async with sem:
            return await send_async(m)
    return await asyncio.gather(*[_send(m) for m in messages], return_exceptions=True)

Il semaforo limita le connessioni SMTP parallele: troppe e il provider applica rate limit (421 4.7.28). Concurrency 10 è un valore safe per la maggior parte dei provider; verificare la doc specifica.

Connection pooling

Aprire e chiudere una connessione SMTP per ogni messaggio in batch è inefficiente (handshake TLS ~150ms). Riusare la connessione:

async def send_batch_pooled(messages: list[EmailMessage]):
    ctx = ssl.create_default_context()
    async with aiosmtplib.SMTP(
        hostname="smtp.targetsmtp.it",
        port=587,
        start_tls=True,
        tls_context=ctx,
        timeout=15,
    ) as smtp:
        await smtp.login("user", "pass")
        for msg in messages:
            try:
                await smtp.send_message(msg)
            except aiosmtplib.SMTPException as e:
                # log, eventuale retry pianificato altrove
                pass

💡 Suggerimento: alcuni provider chiudono la connessione dopo N messaggi (tipicamente 100-500). Wrappare il blocco in try/except per riconnettersi automaticamente, oppure ciclare in chunk di 100 messaggi con connessione fresca per chunk.

Logging strutturato

import logging
import time
import structlog

structlog.configure(processors=[
    structlog.processors.add_log_level,
    structlog.processors.TimeStamper(fmt="iso"),
    structlog.processors.JSONRenderer(),
])
log = structlog.get_logger()

def send_with_log(msg: EmailMessage, key: str):
    bound = log.bind(idempotency_key=key, recipient=msg["To"], subject=msg["Subject"])
    start = time.monotonic()
    try:
        send_email(msg, "smtp.targetsmtp.it", 587, "user", "pass")
        bound.info("email_sent", duration_ms=round((time.monotonic()-start)*1000))
    except smtplib.SMTPException as e:
        bound.error(
            "email_failed",
            error_type=type(e).__name__,
            smtp_code=getattr(e, "smtp_code", None),
            duration_ms=round((time.monotonic()-start)*1000),
        )
        raise

Multipart e allegati

from email.message import EmailMessage
from pathlib import Path

msg = EmailMessage()
msg["From"] = "noreply@targetsmtp.it"
msg["To"] = "cliente@example.it"
msg["Subject"] = "Fattura #4287"

# Text + HTML alternative
msg.set_content("Versione testo: fattura allegata in PDF.")
msg.add_alternative(
    "<p>Versione <strong>HTML</strong>: fattura allegata.</p>",
    subtype="html",
)

# Attachment
pdf = Path("/tmp/fattura-4287.pdf").read_bytes()
msg.add_attachment(pdf, maintype="application", subtype="pdf", filename="fattura-4287.pdf")

Il modulo email di Python 3.6+ ha API moderne e safe. Per allegati > 10MB considera compressione o link a CDN: alcuni provider rifiutano payload > 25MB con 552 5.2.3.

Header obbligatori per deliverability

msg["Message-ID"] = "<order-4287-confirm@targetsmtp.it>"
msg["Date"] = email.utils.formatdate(localtime=True)
msg["List-Unsubscribe"] = (
    "<mailto:unsub@targetsmtp.it?subject=u-4287>, "
    "<https://targetsmtp.it/u/4287>"
)
msg["List-Unsubscribe-Post"] = "List-Unsubscribe=One-Click"
msg["Feedback-ID"] = "orders:transactional:targetsmtp:001"

Senza Message-ID alcuni mailer assegnano valori temporanei meno tracciabili. Senza Date (in RFC 5322 format), alcuni filtri penalizzano. List-Unsubscribe è obbligatorio per bulk sender Gmail dal 2024.

Errori comuni

1. Reuse di smtplib.SMTP tra thread: smtplib NON è thread-safe. Una connessione per thread o lock esplicito
2. Connessione non chiusa su eccezione: usa SEMPRE with statement o try/finally
3. send_message vs sendmail: send_message(msg) usa header del MIME object, sendmail(from, to, msg) sovrascrive envelope. Confondere causa SPF break
4. Encoding UTF-8 mancante: per body con caratteri non-ASCII (italiano, emoji), assicurarsi di settare charset esplicito
5. Login dopo STARTTLS dimenticato: senza login il server rifiuta con 530 5.7.0 Authentication required

Riferimenti

• Python smtplib docs
• aiosmtplib docs
• email.message API
• RFC 8058 — One-Click List-Unsubscribe

Python in produzione vuole soprattutto attenzione a timeout, SSL context corretti e idempotency esplicita. Una volta che il client è solido, la maggior parte degli incidenti viene dal lato server: rate limit, blacklist temporanee, transitori 4xx. Target SMTP fornisce SDK Python ufficiale (in alternativa al raw smtplib) che incapsula questi pattern by default e supporta header Idempotency-Key server-side; il Send-Time Firewall blocca destinatari in suppression o pattern di payload anomali prima che lo script Python consumi un quota di rate limit inutilmente.