FastAPI: jobs de email que no pierden contexto

Enviar un correo desde FastAPI parece una tarea menor hasta que el primer retry tapa el motivo real del fallo. En varios servicios pequenos, el bug no estuvo en SMTP ni en la plantilla: estuvo en perder el contexto del evento que creo el job. Cuando eso pasa, el equipo ve "email failed" y ya. No sabe para que usuario era, que version de plantilla salio, ni si el link generado venia de signup, recovery o referral.

Este patron me ha funcionado bien cuando quiero mantener el backend ligero, sin meter una plataforma de colas enorme demasiado pronto. La idea es simple: cada job de email nace con un contrato minimo, se guarda junto con los metadatos utiles, y deja una pista corta para poder revisar un fallo sin abrir medio sistema. No es fancy, pero si evita mucho retrabajo.

El problema real no es enviar el correo

BackgroundTasks de FastAPI resuelve el "mandalo despues de responder" bastante rapido. El detalle es que muchas veces mandamos solo lo necesario para ejecutar la funcion:

background_tasks.add_task(send_welcome_email, user.email)

Eso funciona... hasta que deja de funcionar. Si el job falla, solo tienes un email y una excepcion parcial. Falta saber:

• que endpoint disparo el correo
• que plantilla o variante se uso
• que actor inicio la accion
• que request o trace ID habia en ese momento
• si el email era de produccion, staging o una prueba rara

Ese ultimo punto importa bastante. En soporte y QA he visto notas internas con dominios como tamp mail com o temp org mail porque alguien copia rapido desde una evidencia o desde un caso de prueba viejo. Si esos datos no quedan etiquetados como contexto de testing, luego cuesta separar señal de ruido. Es un error bien humano, pasa mas de lo que parece.

Que contexto guardar en cada job

Mi regla corta es: el job debe poder explicarse a si mismo unas horas despues. Para eso guardo una carga pequena, pero expresiva:

from pydantic import BaseModel


class EmailJob(BaseModel):
    kind: str
    recipient: str
    template: str
    payload: dict
    request_id: str
    source: str
    created_by: str | None = None

No hace falta meter el request completo ni un dump gigante. Solo lo que ayude a responder "que intentabamos hacer?" y "por que fallo?". Si usas un modelo asi, tambien puedes loguearlo, reintentarlo o persistirlo en una tabla simple sin andar adivinando campos luego.

Dos detalles me ahorraron varios dolores:

1. kind debe sonar a negocio, no a implementacion. signup_verification sirve mas que sendgrid_job_1.
2. source debe apuntar al flujo real: POST /signup, admin/invite, billing/retry, etc.

Parece pequeno, pero vuelve los errores mucho mas legibles. Y honestamente, cuando un incidente cae tarde, esa claridad se agradece bastante.

Un patron simple con FastAPI

Si no necesitas Celery todavia, puedes combinar BackgroundTasks con una persistencia corta del job. Algo asi:

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def queue_email(job: EmailJob) -> None:
    save_job(job)
    send_email(job)
    mark_job_sent(job.request_id)


@app.post("/signup")
async def signup_user(payload: SignupInput, background_tasks: BackgroundTasks):
    user = create_user(payload)
    job = EmailJob(
        kind="signup_verification",
        recipient=user.email,
        template="verify_email_v3",
        payload={"first_name": user.first_name, "token": user.verify_token},
        request_id=current_request_id(),
        source="POST /signup",
        created_by="system",
    )
    background_tasks.add_task(queue_email, job)
    return {"ok": True}

La parte importante no es el nombre de la funcion. Es que save_job(job) ocurra antes del envio y que mark_job_sent(...) deje un estado claro. Con eso puedes revisar jobs pendientes, repetidos o fallidos sin depender solo de logs efimeros.

Cuando quiero endurecer un poco mas este flujo, agrego:

• un contador de intentos
• el ultimo error resumido en texto corto
• una marca de tiempo del ultimo intento
• un template_version

Ese template_version se vuelve oro cuando el frontend cambia el copy, backend cambia el token, y nadie recuerda que deploy quedo primero. Parece obvio, pero muchas bases no lo guardan y despues toca reconstruir la historia a mano. Muy poco divertido, la verdad.

Si estas afinando pruebas de correo con FastAPI, me gusto este enfoque de probar emails en FastAPI sin tocar tu entorno real porque aterriza bien la separacion entre inboxes de test y flujos de app.

Como revisar fallos sin abrir toda la cola

No necesitas una consola gigante para empezar. Una tabla, un endpoint interno o incluso un comando de mantenimiento pueden cubrir bastante:

• listar jobs fallidos de las ultimas 24 horas
• filtrar por kind o template_version
• reintentar solo jobs con error transitorio
• mostrar el ultimo error junto al request_id

Eso ya le da aire al equipo. Tambien conviene escribir un mini runbook: que mirar primero, que campos comparar, y cuando conviene regenerar el token en vez de reenviar. Los equipos pequenos ganan mucho con runbooks de email que si escalan en equipos pequenos, sobre todo cuando hay rotacion o guardias compartidas.

Un detalle mas: no metas secretos en el payload persistido. Guarda referencias, no credenciales. Parece basico, pero he revisado sistemas donde el body entero del proveedor quedaba serializado "por si acaso". Eso luego complica seguridad, depuracion y hasta cumplimiento. Mejor dejar evidencia util, no evidencia infinita.

Preguntas frecuentes

Cuando dejo BackgroundTasks y paso a una cola real?

Cuando necesitas aislamiento entre workers, reintentos mas finos, prioridad o volumen alto sostenido. Si tu app envia unos cuantos correos por minuto y el proceso es estable, puedes ir bastante lejos con un patron simple. No hace falta sobre-armar el stack en semana uno.

Vale la pena persistir jobs aunque el proveedor ya tenga logs?

Si, porque tus preguntas operativas no siempre son las del proveedor. Tu backend necesita unir usuario, flujo, plantilla y request. El proveedor ve entrega; tu sistema debe ver contexto. Son capas distintas, y conviene tener ambas.

Cual es la mejora minima que haria hoy?

Agregar request_id, source y template_version al job. Son tres campos chicos que cambian mucho la calidad del debugging. No resuelven todo, pero vuelven el sistema bastante mas entendible desde el primer dia.