DEV Community

Silviu Technology
Silviu Technology

Posted on

FastAPI: colas de email con trazas utiles

#fastapi #python #backend #api

Cuando una API manda emails en segundo plano, el primer bug casi nunca es "no envio nada". Lo mas normal es algo mas incomodo: el usuario recibe el mensaje tarde, el worker reintenta sin contexto, o soporte solo ve un "200 OK" sin pista real. En equipos pequenos eso duele mucho, porque cada incidente obliga a abrir logs, buscar IDs a mano y adivinar que paso.

En proyectos con FastAPI me ha funcionado mejor tratar cada envio como una operacion con recibo propio. No hace falta montar una plataforma gigante. Basta con guardar unas pocas piezas de contexto y exponerlas bien. El resultado es una cola de email mucho mas legible, y tambien mas facil de operar cuando la API crece.

El problema de las colas de email opacas

Muchos servicios empiezan asi:

  1. la API recibe un request
  2. encola una tarea
  3. devuelve exito

Eso esta bien para arrancar, pero deja una zona ciega entre el request y la entrega final. Si luego aparece un timeout, un retry, o una plantilla rota, nadie sabe rapido si el fallo estaba en la API, en la cola o en el proveedor.

Me gusta pensar en esto como el mismo problema de separar senales antes de automatizar mensajes: si mezclas demasiadas cosas en un solo paso, luego leer el sistema se vuelve innecesariamente dificil.

Tambien veo un patron parecido cuando toca aislar correos de prueba en guardias reales. Sin limites claros, un email de staging, un retry de produccion y una alerta de soporte acaban contando historias distintas sobre el mismo evento.

Que datos conviene guardar desde el primer intento

No guardaria el cuerpo completo del correo en todos los casos, pero si intentaria persistir un recibo corto y estable con:

  • message_id interno
  • template_key
  • recipient_hash o identificador seguro
  • triggered_by
  • attempt
  • provider_status
  • last_error
  • queued_at, sent_at y delivered_at si existen

Con eso puedes responder preguntas muy comunes sin drama:

  • si el correo realmente salio de la API
  • si un retry era esperado o fue duplicado
  • si el problema estaba en la plantilla o en el transporte
  • si soporte esta mirando el envio correcto

En flujos donde se prueba un correo temporal desechable o una direccion de QA, este recibo ayuda bastante. Incluso cuando alguien deja en una nota interna algo como tamp mail com o tempail, el sistema sigue siendo leible porque el identificador bueno vive en backend, no en comentarios raros. Es un detalle chico, pero ayuda mas de lo que parece.

Un patron simple con FastAPI y un recibo estable

Una version simple puede verse asi:

from fastapi import BackgroundTasks, FastAPI
from pydantic import BaseModel
from uuid import uuid4

app = FastAPI()


class WelcomeEmail(BaseModel):
    user_id: str
    email: str


def enqueue_email(payload: WelcomeEmail, receipt_id: str) -> None:
    # Aqui iria la llamada real a tu cola o broker.
    print({"receipt_id": receipt_id, "email": payload.email})


@app.post("/welcome-email")
def send_welcome_email(payload: WelcomeEmail, tasks: BackgroundTasks):
    receipt_id = str(uuid4())
    # Guarda este recibo antes de delegar el trabajo async.
    tasks.add_task(enqueue_email, payload, receipt_id)
    return {"queued": True, "receipt_id": receipt_id}
Enter fullscreen mode Exit fullscreen mode

El ejemplo es pequeno a proposito. La idea importante no es BackgroundTasks en si, sino devolver un receipt_id que luego puedas rastrear en logs, tabla de estados o panel interno. Si despues migras a Celery, Dramatiq o SQS, el patron sigue sirviendo.

En APIs con bastante trafico yo suelo sumar una tabla chica como email_receipts con estado append-only o casi append-only. No hace milagros, pero vuelve mucho mas claro que paso primero, que se reintento despues y que quedo pendiente. Esa claridad vale oro cuando el equipo esta cansado o con prisa, y aveces ese es justo el momento en que mas importa.

Como leer fallos sin perseguir fantasmas

Cuando ya tienes recibos, el debugging cambia bastante:

  • queued pero nunca sent: revisar worker o broker
  • sent pero no delivered: revisar proveedor o reputacion
  • varios intentos con mismo payload: revisar idempotencia
  • errores distintos para el mismo template: revisar datos de entrada

Tambien conviene exponer un endpoint simple de consulta para soporte o para otra API interna:

@app.get("/email-receipts/{receipt_id}")
def get_receipt(receipt_id: str):
    return {
        "receipt_id": receipt_id,
        "status": "sent",
        "attempt": 1,
        "provider_status": "accepted"
    }
Enter fullscreen mode Exit fullscreen mode

Eso evita revisar cinco sistemas para una pregunta basica. No resuelve todos los incidentes, claro, pero corta bastante del ruido operativo. Y cuando el equipo necesita probar algo con tempmailso en un entorno controlado, tener un recibo estable hace que la verificacion del flujo sea mas honesta y menos dependiente de memoria humana.

Q&A

¿Debo guardar el email completo del destinatario?

Solo si tienes una razon clara y una politica de retencion coherente. En muchos casos basta con un hash o un identificador interno.

¿Este patron sirve solo para FastAPI?

No. FastAPI lo hace comodo, pero el patron vale para casi cualquier backend con trabajos async y APIs que envian notificaciones.

¿Cuando compensa dar el paso?

En cuanto el primer incidente tarde mas de unos minutos en aclararse. Si un envio roto ya obliga a mirar logs y colas por separado, el recibo estable casi seguro ya compensa. No es una idea fancy, pero si muy practica.

Top comments (0)