Verifica delle email in Python

Questo tutorial ripercorre l'integrazione della verifica delle email di Mailbeam in un'applicazione web Python. Vedrai esempi sia per FastAPI (asincrono, consigliato nei progetti nuovi) sia per Flask (sincrono, per applicazioni già esistenti).

Che cosa costruirai

Una dipendenza validate_email riutilizzabile per FastAPI
Un helper sincrono verify_email per Flask o altri framework sincroni
Una gestione degli errori che non blocca gli utenti se l'API va in errore

Prerequisiti

Python 3.9 o successivo
Un ambiente virtuale (venv, poetry o conda)
Una chiave API di Mailbeam (iscriviti gratis)

Passo 1 — Aggiungi un piccolo client

Non c'è nessun SDK da installare. Mailbeam è un unico endpoint HTTP, quindi un modulo breve nel tuo codice lo copre e non c'è niente da tenere aggiornato. Gli SDK ufficiali sono nella roadmap; questo è quello che si fa oggi.

pip install httpx        # oppure usa requests, se è già nel progetto

# app/mailbeam.py
import os

import httpx

ENDPOINT = "https://api.mailbeam.dev/v1/verify"
HEADERS = {"Authorization": f"Bearer {os.environ['MAILBEAM_KEY']}"}


class MailbeamError(Exception):
    """Qualsiasi cosa ci abbia impedito di ottenere un esito."""


async def verify_email(email: str, timeout: float = 3.0) -> dict:
    try:
        async with httpx.AsyncClient(timeout=timeout) as client:
            response = await client.post(ENDPOINT, json={"email": email}, headers=HEADERS)
            response.raise_for_status()
            return response.json()
    except httpx.HTTPError as exc:
        raise MailbeamError(str(exc)) from exc


def verify_email_sync(email: str, timeout: float = 3.0) -> dict:
    try:
        response = httpx.post(ENDPOINT, json={"email": email}, headers=HEADERS, timeout=timeout)
        response.raise_for_status()
        return response.json()
    except httpx.HTTPError as exc:
        raise MailbeamError(str(exc)) from exc

Passo 2 — Definisci la tua variabile d'ambiente

# .env
MAILBEAM_KEY=mb_live_xxxxxxxxxxxxxxxxxxxx

Caricala con python-dotenv o con il metodo che preferisce il tuo framework:

import os
from dotenv import load_dotenv

load_dotenv()
MAILBEAM_KEY = os.environ["MAILBEAM_KEY"]

Passo 3 — FastAPI: crea la dipendenza di validazione

# app/dependencies.py
import logging

from fastapi import HTTPException

from app.mailbeam import MailbeamError, verify_email


async def validate_email(email: str) -> str:
    """
    Dipendenza di FastAPI che verifica un indirizzo email.
    Solleva un HTTP 422 se l'email non è valida o è di bassa qualità.
    In caso di errore dell'API di Mailbeam lascia passare.
    """
    try:
        result = await verify_email(email)
    except MailbeamError as exc:
        # Registra l'errore ma non bloccare: il nostro guasto non è un problema dell'utente.
        logging.error("Verifica di Mailbeam fallita: %s", exc)
        return email

    if not result["valid"] or result["score"] < 60:
        raise HTTPException(
            status_code=422,
            detail={
                "error": "Inserisci un indirizzo email valido.",
                "code": result["reason"] or "invalid_email",
            },
        )

    return email

Passo 4 — Aggiungila al tuo endpoint di registrazione

# app/routes/auth.py
from fastapi import APIRouter, Depends
from pydantic import BaseModel, EmailStr
from typing import Annotated
from app.dependencies import validate_email

router = APIRouter()


class SignupRequest(BaseModel):
    email: EmailStr
    password: str


@router.post("/signup", status_code=201)
async def signup(
    request: SignupRequest,
    # La dipendenza viene eseguita PRIMA del gestore e solleva 422 se non è valida
    verified_email: Annotated[str, Depends(validate_email)],
):
    user = await create_user(email=verified_email, password=request.password)
    return {"user": user}

La tua applicazione principale:

# app/main.py
from fastapi import FastAPI
from app.routes.auth import router

app = FastAPI()
app.include_router(router, prefix="/api/auth")

Passo 5 — Versione per Flask (sincrona)

# app/utils/email.py
import logging

from app.mailbeam import MailbeamError, verify_email_sync


def check_email(email: str) -> tuple[bool, str | None]:
    """
    Restituisce (è_valida, codice_del_motivo).
    In caso di errore dell'API lascia passare (restituisce True).
    """
    try:
        result = verify_email_sync(email)
    except MailbeamError as exc:
        logging.error("Verifica di Mailbeam fallita: %s", exc)
        return True, None  # lascia passare

    if not result["valid"] or result["score"] < 60:
        return False, result["reason"] or "invalid_email"
    return True, None


# app/routes/auth.py (Flask)
from flask import Blueprint, request, jsonify
from app.utils.email import check_email

bp = Blueprint("auth", __name__)


@bp.post("/api/auth/signup")
def signup():
    data = request.get_json()
    email = data.get("email", "")

    is_valid, reason = verify_email_sync(email)
    if not is_valid:
        return jsonify({"error": "Inserisci un'email valida.", "code": reason}), 422

    user = create_user(email=email, password=data.get("password"))
    return jsonify({"user": user}), 201

Provare l'integrazione

# tests/test_signup.py
import pytest
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)


def test_signup_with_valid_email():
    # Usa il dominio di prova di Mailbeam: restituisce sempre valido e non consuma quota
    response = client.post(
        "/api/auth/signup",
        json={"email": "user@valid.mailbeam-test.dev", "password": "pass1234"},
    )
    assert response.status_code == 201


def test_signup_with_invalid_email():
    response = client.post(
        "/api/auth/signup",
        json={"email": "user@invalid.mailbeam-test.dev", "password": "pass1234"},
    )
    assert response.status_code == 422
    assert response.json()["detail"]["code"] is not None


def test_signup_with_disposable_email():
    response = client.post(
        "/api/auth/signup",
        json={"email": "temp@disposable.mailbeam-test.dev", "password": "pass1234"},
    )
    assert response.status_code == 422

Aggiungere uno strato di cache

# app/dependencies.py
from functools import lru_cache
import asyncio

# Cache TTL essenziale nel processo con un dict (in produzione, usa Redis)
_cache: dict[str, tuple] = {}

async def validate_email(email: str) -> str:
    normalized = email.lower().strip()

    if normalized in _cache:
        valid, score, reason = _cache[normalized]
        if not valid or score < 60:
            raise HTTPException(
                status_code=422,
                detail={"error": "Inserisci un'email valida.", "code": reason},
            )
        return email

    try:
        result = await verify_email(normalized)
    except MailbeamError:
        return email  # lascia passare

    _cache[normalized] = (result["valid"], result["score"], result["reason"])

    if not result["valid"] or result["score"] < 60:
        raise HTTPException(
            status_code=422,
            detail={"error": "Inserisci un'email valida.", "code": result["reason"]},
        )

    return email

Buone pratiche

Pratica	Perché
Usare il client asincrono con FastAPI	Non blocca il ciclo degli eventi
Usare il client sincrono con Flask	Nessun peso asincrono nei framework sincroni
Lasciar passare davanti a `MailbeamError`	Un guasto dell'API non deve rompere la registrazione
Normalizzare l'email prima di metterla in cache	Evita mancati riscontri per differenze di maiuscole
Usare i domini di prova in pytest	Test isolati e senza consumo di quota

Lista di controllo per la produzione

MAILBEAM_KEY nei segreti d'ambiente, non nel codice
Registrazione degli errori configurata (logging.error, non print)
Cache configurata (Redis nelle pubblicazioni con più worker)
Domini di prova usati nelle fixture di pytest
Il messaggio di errore 422 è mostrato in modo comprensibile nel tuo frontend