Introducción al Diseño Práctico de una Base de Notificaciones y Envío de Correos con FastAPI: Patrones de Trabajo con BackgroundTasks, Colas de Jobs, Plantillas, Reintentos y Auditoría

Resumen

• Las notificaciones y el envío de correos son una base importante para muchos procesos de negocio, como registro de usuarios, respuestas a consultas, facturación y alertas de SLA. En FastAPI, conviene diferenciar entre BackgroundTasks para procesos ligeros posteriores y colas de jobs como Celery cuando se requiere fiabilidad o reintentos.
• El envío de correos puede hacerse directamente mediante SMTP o mediante APIs externas de distribución de correo. aiosmtplib, cliente SMTP asíncrono para Python, presenta la corrutina send() como entrada principal para el envío.
• En la práctica, no basta con construir el cuerpo del correo en el momento y enviarlo. Es importante diseñarlo como una “base de notificaciones”, incluyendo plantillas, historial de envío, reintentos, prevención de duplicados, fallos de entrega, cancelación de suscripción, logs de auditoría y métricas.
• Para notificaciones importantes, es más seguro crear primero un job de notificación y enviarlo en un proceso posterior, sin esperar a que el envío termine dentro de la respuesta de la API. Celery se describe oficialmente como una cola de tareas distribuida para procesamiento en tiempo real y programación.
• Este artículo organiza paso a paso cómo pensar una base de notificaciones y correos en FastAPI: clasificación de usos, diseño de plantillas, elección entre síncrono/asíncrono, patrón Outbox, reintentos, auditoría y pruebas.

Quién se Beneficia de Leer Esto

Personas que desarrollan o aprenden por cuenta propia

Está dirigido a quienes quieren enviar con FastAPI correos de confirmación tras registro, recepción de consultas o restablecimiento de contraseña. Al principio puede funcionar con llamar directamente a send_email(), pero cuando empieza la operación aparecen preguntas como “qué hacer si no se envía”, “cómo evitar enviar el mismo correo dos veces” o “cómo gestionar plantillas”.

En este artículo presentamos un diseño gradual: empezar con BackgroundTasks y avanzar a una cola de jobs cuando sea necesario. Con el sistema de dependencias de FastAPI, es fácil compartir clientes de correo y configuraciones.

Personas backend en equipos pequeños

Está dirigido a equipos donde aumentan respuestas a consultas, notificaciones de facturación, correos masivos desde panel de administración o notificaciones internas, y el envío de correos empieza a dispersarse por routers y capas de servicio.

Al centralizar la base de notificaciones, las plantillas, el historial de envío, los reintentos, los logs de auditoría y la gestión de fallos quedan en un solo lugar. También resulta más fácil compartir reglas como “las notificaciones importantes siempre dejan historial”, “solo las notificaciones ligeras usan BackgroundTasks” o “los envíos masivos van a la cola de jobs”.

Equipos SaaS y startups

Está dirigido a equipos que manejan múltiples tipos de notificaciones: correos a clientes, notificaciones dentro del producto, Webhooks, Slack, notificaciones de facturación y alertas de SLA.

En esta etapa, el correo no es una función auxiliar, sino parte central de la experiencia del cliente y la operación. Hay que diseñar retrasos, envíos duplicados, reintentos, bajas de suscripción y trazabilidad de auditoría. Una cola distribuida como Celery encaja bien con una base de notificaciones que incluye procesamiento masivo y programación.

Evaluación de Accesibilidad

• El artículo comienza con un resumen para captar primero la visión general.
• Términos como “correo”, “notificación”, “job”, “plantilla” y “Outbox” se explican brevemente al aparecer.
• Los ejemplos de código son cortos y cada bloque muestra un solo propósito.
• Se separan las etapas de adopción para desarrollo individual, equipos pequeños y equipos SaaS.
• El nivel objetivo es equivalente a AA.

1. Una Base de Notificaciones no es una “Función para Enviar Correos”

Al implementar envío de correos en FastAPI, al principio suele apetecer escribir una función como esta:

def send_email(to: str, subject: str, body: str) -> None:
    ...

Como primer paso, esto puede ser suficiente.
Pero en la práctica aparecen pronto problemas como:

• dónde gestionar el cuerpo del correo
• cómo evitar enviar la misma notificación dos veces
• si se reintenta cuando falla el envío
• qué hacer si el usuario canceló la recepción
• cómo tratar diferencias entre notificaciones importantes, como facturación o restablecimiento de contraseña
• cómo guardar historial de envío
• cómo investigar cuando un cliente dice “no me llegó”

Es decir, conviene diseñar el envío de correos no como una función, sino como una base.

2. Clasificar Primero los Tipos de Notificación

Antes de crear la base, clasifica los tipos de notificación.
Si todo se trata igual, se mezclan notificaciones importantes y ligeras.

2.1 Correos Transaccionales

Son correos directamente ligados a una acción del usuario o a un evento de negocio.

Ejemplos:

• confirmación de correo electrónico
• restablecimiento de contraseña
• recepción de consulta
• respuesta de atención al cliente
• emisión de factura
• notificación de fallo de pago
• notificación de cambio de contrato

Tienen alta importancia y suelen necesitar historial de envío y reintentos.

2.2 Correos de Marketing

Son campañas, avisos o newsletters.

Ejemplos:

• presentación de nuevas funciones
• campaña promocional
• newsletter mensual

Aquí son importantes la baja de suscripción, segmentación y control de volumen.

2.3 Notificaciones Internas

Son notificaciones para miembros internos o equipos de operación.

Ejemplos:

• nueva consulta recibida
• ticket urgente
• alerta de fallo de facturación
• fallo de API externa
• alerta de incumplimiento de SLA

Pueden enviarse no solo por correo, sino también a Slack o Webhook.

3. Política Básica en FastAPI: No Escribir el Envío Directamente en el Router

Si llamas directamente a SMTP o a una API externa desde el router, el código se vuelve difícil de mantener.

Una forma que conviene evitar es esta:

@router.post("/signup")
def signup(payload: SignupRequest):
    # Crear usuario
    # Construir el cuerpo del correo aquí
    # Enviar por SMTP aquí
    return {"ok": True}

Es preferible separar responsabilidades:

• Router
  • recibe la solicitud
• Capa de servicio
  • ejecuta procesos de negocio como crear usuarios o responder consultas
• Servicio de notificaciones
  • crea eventos de notificación
• Cliente de correo
  • envía realmente por SMTP o API externa
• Cola de jobs
  • gestiona envíos pesados o reintentos

El sistema de dependencias de FastAPI encaja bien para inyectar estos componentes.

4. Crear un Cliente de Correo Mínimo

Primero, encapsula el envío de correos en una clase dedicada.
Aquí se muestra solo una interfaz pensada para SMTP.

from dataclasses import dataclass

@dataclass
class EmailMessage:
    to: str
    subject: str
    text_body: str
    html_body: str | None = None

class EmailClient:
    async def send(self, message: EmailMessage) -> None:
        ...

Al ocultar la implementación, luego será más fácil cambiar de SMTP a una API externa de correo.

5. Ejemplo de Envío SMTP Asíncrono con aiosmtplib

Si lo combinas con endpoints asíncronos de FastAPI, una opción es usar un cliente SMTP asíncrono. La documentación de aiosmtplib recomienda la corrutina send() como entrada principal para enviar correos.

from email.message import EmailMessage as PyEmailMessage
import aiosmtplib

class SMTPEmailClient:
    def __init__(
        self,
        hostname: str,
        port: int,
        username: str | None = None,
        password: str | None = None,
        use_tls: bool = True,
    ):
        self.hostname = hostname
        self.port = port
        self.username = username
        self.password = password
        self.use_tls = use_tls

    async def send(self, message: EmailMessage) -> None:
        email = PyEmailMessage()
        email["From"] = "no-reply@example.com"
        email["To"] = message.to
        email["Subject"] = message.subject
        email.set_content(message.text_body)

        if message.html_body:
            email.add_alternative(message.html_body, subtype="html")

        await aiosmtplib.send(
            email,
            hostname=self.hostname,
            port=self.port,
            username=self.username,
            password=self.password,
            use_tls=self.use_tls,
        )

Si esta implementación no se escribe directamente en el router, sino que se llama desde el servicio de notificaciones, será más fácil mantenerla.

6. Diseño de Plantillas: No Incrustar el Cuerpo del Correo en el Código

Si escribes el cuerpo del correo directamente en Python, tendrás que tocar código cada vez que cambie el texto.
A medida que aumentan las notificaciones, es más manejable separar plantillas.

6.1 Definir Claves de Plantilla

Ejemplos:

• user.email_verification
• user.password_reset
• ticket.created
• ticket.replied
• billing.payment_failed
• subscription.canceled

Usar el formato <domain>.<event> ayuda a mantener el orden.

6.2 Definir Modelos de Entrada para Plantillas

from pydantic import BaseModel, EmailStr

class TicketReplyEmailContext(BaseModel):
    customer_email: EmailStr
    ticket_id: int
    subject: str
    reply_body: str

Modelar los valores que se pasan a la plantilla aclara qué variables son necesarias.
También reduce errores como que la plantilla necesite reply_body y el llamador olvide pasarlo.

7. Crear un Servicio de Notificaciones

En lugar de llamar directamente al cliente de correo, introduce un servicio de notificaciones.

class NotificationService:
    def __init__(self, email_client: EmailClient):
        self.email_client = email_client

    async def send_ticket_reply_email(self, context: TicketReplyEmailContext) -> None:
        message = EmailMessage(
            to=context.customer_email,
            subject=f"Re: {context.subject}",
            text_body=context.reply_body,
            html_body=f"<p>{context.reply_body}</p>",
        )
        await self.email_client.send(message)

Con esta forma, será más fácil reunir en el servicio de notificaciones:

• selección de plantillas
• comprobación de baja de suscripción
• guardado de historial
• registro de métricas
• logs de auditoría

8. Inyectar el Servicio de Notificaciones con Dependencias de FastAPI

En FastAPI, es práctico construir el servicio de notificaciones mediante funciones de dependencia.

from fastapi import Depends

def get_email_client() -> EmailClient:
    return SMTPEmailClient(
        hostname="smtp.example.com",
        port=465,
        username="smtp-user",
        password="smtp-password",
    )

def get_notification_service(
    email_client: EmailClient = Depends(get_email_client),
) -> NotificationService:
    return NotificationService(email_client=email_client)

En el router se puede usar así:

from fastapi import APIRouter, Depends

router = APIRouter(prefix="/admin/tickets", tags=["admin-tickets"])

@router.post("/{ticket_id}/reply")
async def reply_ticket(
    ticket_id: int,
    notification_service: NotificationService = Depends(get_notification_service),
):
    context = TicketReplyEmailContext(
        customer_email="customer@example.com",
        ticket_id=ticket_id,
        subject="Sobre su consulta",
        reply_body="Gracias por contactarnos.",
    )
    await notification_service.send_ticket_reply_email(context)
    return {"status": "sent"}

Sin embargo, de esta forma la respuesta espera hasta que termine el envío.
Incluso para notificaciones ligeras, muchas veces conviene moverlas a un proceso posterior por experiencia de usuario.

9. Mover Notificaciones Ligeras Después de la Respuesta con BackgroundTasks

FastAPI explica que BackgroundTasks permite definir procesos que se ejecutan después de enviar la respuesta. Es adecuado para tareas que el cliente no necesita esperar, como notificaciones por correo o envío de logs.

from fastapi import BackgroundTasks

def send_ticket_reply_email_sync(ticket_id: int, email: str) -> None:
    # En la práctica, aquí iría el envío de correo
    pass

@router.post("/{ticket_id}/reply")
def reply_ticket(
    ticket_id: int,
    background_tasks: BackgroundTasks,
):
    # Guardar mensaje de respuesta en DB

    background_tasks.add_task(
        send_ticket_reply_email_sync,
        ticket_id,
        "customer@example.com",
    )

    return {"status": "reply_saved"}

BackgroundTasks es muy útil, pero hay que tener cuidado con notificaciones que requieren fiabilidad.
Como se ejecuta dentro del mismo proceso, es débil ante caídas del proceso o grandes volúmenes. Para notificaciones importantes o que requieren reintentos, conviene considerar una cola de jobs.

10. Gestionar Notificaciones Importantes con el Patrón Outbox

El patrón Outbox consiste en guardar primero en la base de datos la notificación que debe enviarse y luego hacer que un worker la envíe.
Así es más fácil mantener consistencia entre el proceso de negocio y el envío de correos.

10.1 Imagen de una Tabla Outbox

from datetime import datetime
from pydantic import BaseModel
from typing import Literal

NotificationStatus = Literal["pending", "sending", "sent", "failed"]

class NotificationOutboxItem(BaseModel):
    id: int
    type: str
    recipient: str
    payload: dict
    status: NotificationStatus
    retry_count: int = 0
    created_at: datetime
    sent_at: datetime | None = None

10.2 En el Proceso de Negocio no se “Envía”, se “Encola”

def enqueue_ticket_reply_notification(ticket_id: int, customer_email: str, reply_body: str) -> None:
    # En la práctica sería un INSERT en DB
    outbox_item = {
        "type": "ticket.replied",
        "recipient": customer_email,
        "payload": {
            "ticket_id": ticket_id,
            "reply_body": reply_body,
        },
        "status": "pending",
    }

Con este diseño, aunque guardar la respuesta tenga éxito y solo falle el envío del correo, luego se puede revisar el Outbox y reenviar.

11. Enviar a una Cola de Jobs como Celery

Cuando hay muchas notificaciones, se quiere reintentar o se necesita rastrear fallos, una cola de jobs es adecuada. Celery se describe oficialmente como un sistema distribuido para procesamiento de mensajes, centrado en procesamiento en tiempo real y programación.

11.1 Procesos que Conviene Convertir en Jobs

• notificaciones de fallo de pago
• correos de restablecimiento de contraseña
• notificación de CSV generado
• envío masivo de correos
• alertas de incumplimiento de SLA
• notificaciones de respuesta de soporte
• recordatorios

11.2 Ejemplo de Tarea Celery

from celery import Celery

celery_app = Celery(
    "notifications",
    broker="redis://redis:6379/0",
    backend="redis://redis:6379/1",
)

@celery_app.task(
    name="notifications.send_email",
    autoretry_for=(Exception,),
    retry_kwargs={"max_retries": 3},
)
def send_email_task(outbox_id: int) -> None:
    # Obtener desde DB usando outbox_id y enviar correo
    # Si tiene éxito, status=sent
    # Si falla, dejar que Celery haga retry
    pass

Al combinar Outbox y Celery, resulta más sencillo manejar tanto historial de envío como reintentos.

12. Evitar Envíos Duplicados: Usar Claves de Idempotencia

Uno de los mayores riesgos en notificaciones es enviar el mismo correo varias veces.
En especial en facturación, restablecimiento de contraseña o cambios de contrato, los duplicados generan ansiedad en clientes.

Por eso conviene que las notificaciones tengan una clave de idempotencia.

Ejemplos:

ticket_reply:{ticket_id}:{message_id}
password_reset:{user_id}:{token_id}
payment_failed:{invoice_id}

Si el Outbox tiene dedup_key y una restricción única, será más seguro.

class NotificationOutboxItem(BaseModel):
    id: int
    dedup_key: str
    type: str
    recipient: str
    payload: dict
    status: NotificationStatus

Si ya existe una notificación con el mismo dedup_key, no se crea una nueva.
Así se evita el doble registro incluso con reintentos o eventos duplicados.

13. Pensar Siempre en Bajas de Suscripción y Preferencias de Notificación

En una base de notificaciones, no solo importa “enviar”, sino también diseñar cuándo “no enviar”.

Puede haber preferencias por usuario o tenant, como:

• recibir notificaciones del producto
• recibir notificaciones de facturación
• recibir correos de marketing
• recibir alertas de SLA
• recibir resumen semanal

Sin embargo, no todas las notificaciones pueden detenerse.
Por ejemplo, restablecimiento de contraseña o fallos de pago son correos transaccionales necesarios para mantener la cuenta, y deben tratarse por separado de los correos de marketing.

13.1 Separar Categorías de Notificación

NotificationCategory = Literal[
    "transactional",
    "billing",
    "security",
    "marketing",
    "internal",
]

El tratamiento de bajas cambia según la categoría.

def can_send_notification(category: str, user_settings: dict) -> bool:
    if category in {"transactional", "security"}:
        return True
    if category == "marketing":
        return bool(user_settings.get("marketing_opt_in"))
    return True

Clasificar así facilita trasladar reglas operativas a implementación.

14. Separar Historial de Envío y Logs de Auditoría

En una base de notificaciones, conviene no mezclar historial de envío y auditoría.

Historial de Envío

• qué notificación se intentó enviar
• a quién se envió
• si tuvo éxito o falló
• cuántos reintentos hubo
• ID de mensaje del servicio externo de correo

Log de Auditoría

• por operación de quién se generó la notificación
• en qué evento de negocio se basó
• si debe conservarse como operación importante

Por ejemplo, en un correo de respuesta de soporte:

• Historial de envío: historial del correo enviado al cliente
• Auditoría: historial de que un agente respondió al ticket

Guardar ambos facilita mucho la atención a consultas y la investigación de incidentes.

15. No Incluir Demasiada Información Personal o Confidencial en el Correo

Una vez enviado, un correo no puede recuperarse.
Además, puede reenviarse o capturarse fácilmente.

Por eso es mejor no incluir demasiada información confidencial en el cuerpo.

Conviene evitar:

• la contraseña en sí
• tokens válidos durante mucho tiempo
• información completa y detallada de facturación
• exceso de datos personales
• notas internas

En correos de restablecimiento de contraseña, es más seguro enviar una URL válida por poco tiempo y realizar los detalles del proceso tras iniciar sesión.

16. Versionado de Plantillas de Notificación

Las plantillas cambian con el tiempo.

• cambiar asunto
• mejorar redacción
• añadir pie de página
• incluir texto legal
• soportar varios idiomas

Por eso, en notificaciones importantes, es útil guardar en el historial con qué versión de plantilla se envió.

class NotificationOutboxItem(BaseModel):
    id: int
    type: str
    template_key: str
    template_version: str
    recipient: str
    payload: dict
    status: NotificationStatus

Esto permite rastrear posteriormente qué texto contenía un correo en cierta época.

17. Diseño de Plantillas Pensando en Multilenguaje

En SaaS, puede ser necesario enviar notificaciones en varios idiomas, como japonés e inglés.
En ese caso, es más ordenado separar la clave de plantilla y el locale, en lugar de mezclar el idioma en la clave.

template_key = "ticket.replied"
locale = "ja"

En la implementación se puede resolver así:

def resolve_template(template_key: str, locale: str) -> str:
    return f"templates/{locale}/{template_key}.html"

18. Tomar Métricas de Notificaciones

En notificaciones, es muy importante saber si se enviaron o no.
Como mínimo, conviene poder ver estas métricas:

• número de notificaciones creadas
• número de envíos exitosos
• número de fallos de envío
• número de reintentos
• cantidad acumulada en Outbox
• retraso medio de envío
• número de envíos por categoría

Por ejemplo, si el Outbox tiene created_at y sent_at, se puede calcular el retraso:

delay_seconds = (sent_at - created_at).total_seconds()

Las notificaciones están directamente conectadas con la experiencia de usuario.
Hay que detectar rápidamente cuando “no se están enviando”.

19. Política de Pruebas: Verificar sin Enviar Correos Reales

En pruebas, es más seguro no llamar a SMTP real ni a APIs externas de correo.

19.1 Usar un FakeEmailClient

class FakeEmailClient:
    def __init__(self):
        self.sent: list[EmailMessage] = []

    async def send(self, message: EmailMessage) -> None:
        self.sent.append(message)

Si se sustituye mediante dependencias, se puede verificar “qué se intentó enviar” sin enviarlo realmente.

async def test_send_ticket_reply_email():
    fake = FakeEmailClient()
    service = NotificationService(email_client=fake)

    context = TicketReplyEmailContext(
        customer_email="customer@example.com",
        ticket_id=1,
        subject="Consulta",
        reply_body="Cuerpo de la respuesta",
    )

    await service.send_ticket_reply_email(context)

    assert len(fake.sent) == 1
    assert fake.sent[0].to == "customer@example.com"

19.2 Pruebas del Outbox

Si usas Outbox, son importantes pruebas como:

• un evento de notificación crea un registro Outbox
• no se crean duplicados con el mismo dedup_key
• al enviar con éxito pasa a status=sent
• al fallar aumenta retry_count
• si existe baja de suscripción, no se crea o queda como skipped

20. Patrones de Fallo Frecuentes

20.1 Enviar Correos Directamente desde el Router

Al principio es fácil, pero cuando entran plantillas, bajas, historial y reintentos, se vuelve frágil. Es más seguro llevarlo al servicio de notificaciones.

20.2 Resolver Notificaciones Importantes Solo con BackgroundTasks

BackgroundTasks es útil para procesar después de la respuesta, pero es débil ante caídas del proceso y grandes volúmenes. Para notificaciones importantes, considera Outbox o colas de jobs.

20.3 Incrustar el Texto de Plantillas en el Código

Cada cambio de redacción exige modificar código. Es recomendable separar clave de plantilla y contexto.

20.4 No Considerar Bajas de Suscripción

Si no separas correos de marketing y transaccionales, pueden aparecer problemas legales, operativos y de experiencia de cliente.

20.5 No Guardar Historial de Envío

Cuando alguien dice “no llegó”, no podrás investigar. Las notificaciones importantes deben tener historial y estado.

21. Roadmap por Tipo de Lector

Personas que Desarrollan o Aprenden por Cuenta Propia

1. Crear una clase EmailClient
2. Crear NotificationService
3. Mover notificaciones ligeras después de la respuesta con BackgroundTasks
4. Introducir claves de plantilla
5. Probar con un cliente fake

Personas Ingenieras en Equipos Pequeños

1. Inventariar tipos de notificación
2. Separar correos transaccionales y de marketing
3. Introducir una tabla Outbox
4. Enviar notificaciones importantes a una cola de jobs
5. Diseñar por separado historial de envío y logs de auditoría

Equipos SaaS y Startups

1. Organizar la base de notificaciones como base de eventos de negocio
2. Gestionar plantillas, locales y versiones
3. Evitar duplicados con claves de idempotencia
4. Monitorizar acumulación en Outbox y tasa de fallo
5. Integrar soporte, facturación, SLA y Webhooks

Enlaces de Referencia

• FastAPI

  • Background Tasks
  • BackgroundTasks Reference
  • Dependencies

• aiosmtplib

  • The send Coroutine – aiosmtplib

• Celery

  • Celery – Distributed Task Queue

Conclusión

• Las notificaciones y el envío de correos suelen verse como algo ligero dentro de una app FastAPI, pero en la práctica están profundamente relacionados con experiencia de cliente, facturación, soporte, auditoría y seguridad.
• Está bien empezar moviendo notificaciones ligeras después de la respuesta con BackgroundTasks. Sin embargo, para notificaciones importantes o que requieren reintentos, es más seguro avanzar hacia Outbox o colas de jobs como Celery.
• Conviene convertir el cuerpo del correo en plantillas y gestionar por separado tipo de notificación, categoría, versión de plantilla y locale, para resistir mejor cambios futuros.
• Las claves de idempotencia son efectivas para evitar duplicados. Si el Outbox tiene restricciones únicas, será más resistente ante reintentos y eventos dobles.
• Al organizar historial de envío, logs de auditoría y métricas, la base de notificaciones permite rastrear “si se envió”, “por qué falló” y “por operación de quién se generó”.

Como próximos artículos, encajan naturalmente “Diseño de una API de Gestión de SLA y Escalación con FastAPI” o “Introducción Práctica a la Arquitectura Orientada a Eventos con FastAPI”.