Verificación de email en Python

Este tutorial recorre la integración de la verificación de email de Mailbeam en una aplicación web de Python. Verás ejemplos tanto para FastAPI (asíncrono, recomendado en proyectos nuevos) como para Flask (síncrono, para aplicaciones ya existentes).

Qué vas a construir

Una dependencia validate_email reutilizable para FastAPI
Un ayudante síncrono verify_email para Flask u otros frameworks síncronos
Una gestión de errores que no bloquea a los usuarios si la API falla

Requisitos previos

Python 3.9 o posterior
Un entorno virtual (venv, poetry o conda)
Una clave de API de Mailbeam (regístrate gratis)

Paso 1 — Añade un cliente pequeño

No hay ningún SDK que instalar. Mailbeam es un único endpoint HTTP, así que un módulo corto en tu propio código lo cubre y no hay nada que mantener actualizado. Los SDK oficiales están en la hoja de ruta; esto es lo que hay que hacer hoy.

pip install httpx        # o usa requests, si ya está en el proyecto

# 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):
    """Cualquier cosa que nos haya impedido obtener un veredicto."""


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

Paso 2 — Define tu variable de entorno

# .env
MAILBEAM_KEY=mb_live_xxxxxxxxxxxxxxxxxxxx

Cárgala con python-dotenv o con el método que prefiera tu framework:

import os
from dotenv import load_dotenv

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

Paso 3 — FastAPI: crea la dependencia de validación

# app/dependencies.py
import logging

from fastapi import HTTPException

from app.mailbeam import MailbeamError, verify_email


async def validate_email(email: str) -> str:
    """
    Dependencia de FastAPI que verifica una dirección de email.
    Lanza un HTTP 422 si el email es inválido o de baja calidad.
    Falla en abierto (deja pasar) ante errores de la API de Mailbeam.
    """
    try:
        result = await verify_email(email)
    except MailbeamError as exc:
        # Registra el error pero no bloquees: nuestra caída no es problema del usuario.
        logging.error("Ha fallado la verificación de Mailbeam: %s", exc)
        return email

    if not result["valid"] or result["score"] < 60:
        raise HTTPException(
            status_code=422,
            detail={
                "error": "Escribe una dirección de email válida.",
                "code": result["reason"] or "invalid_email",
            },
        )

    return email

Paso 4 — Añádela a tu endpoint de registro

# 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 dependencia se ejecuta ANTES del handler y lanza 422 si no es válido
    verified_email: Annotated[str, Depends(validate_email)],
):
    user = await create_user(email=verified_email, password=request.password)
    return {"user": user}

Tu aplicación principal:

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

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

Paso 5 — Versión para Flask (síncrona)

# app/utils/email.py
import logging

from app.mailbeam import MailbeamError, verify_email_sync


def check_email(email: str) -> tuple[bool, str | None]:
    """
    Devuelve (es_valido, codigo_de_motivo).
    Falla en abierto (devuelve True) ante errores de la API.
    """
    try:
        result = verify_email_sync(email)
    except MailbeamError as exc:
        logging.error("Ha fallado la verificación de Mailbeam: %s", exc)
        return True, None  # falla en abierto

    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": "Escribe un email válido.", "code": reason}), 422

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

Probar la integración

# 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 el dominio de prueba de Mailbeam: siempre devuelve válido y no consume cuota
    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

Añadir una capa de caché

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

# Caché TTL sencilla en el proceso con un dict (en producción, 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": "Escribe un email válido.", "code": reason},
            )
        return email

    try:
        result = await verify_email(normalized)
    except MailbeamError:
        return email  # falla en abierto

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

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

    return email

Buenas prácticas

Práctica	Por qué
Usar el cliente asíncrono con FastAPI	No bloquea el bucle de eventos
Usar el cliente síncrono con Flask	Sin sobrecarga asíncrona en frameworks síncronos
Fallar en abierto ante `MailbeamError`	Una caída de la API no debería romper el registro
Normalizar el email antes de cachear	Evita fallos de caché por diferencias de mayúsculas
Usar dominios de prueba en pytest	Tests aislados y sin consumo de cuota

Checklist de producción

MAILBEAM_KEY en los secretos del entorno, no en el código
Registro de errores configurado (logging.error, no print)
Caché configurada (Redis en despliegues con varios workers)
Dominios de prueba usados en las fixtures de pytest
El mensaje de error 422 se muestra de forma comprensible en tu frontend