Background tasks

Crear tareas en background con BackgroundTasks

Las tareas en background permiten ejecutar operaciones que no requieren que el cliente espere una respuesta inmediata. En lugar de hacer que el usuario aguarde mientras se procesa un archivo o se envía un email, podemos devolver la respuesta HTTP de forma inmediata y ejecutar estas operaciones en segundo plano.

FastAPI proporciona la clase BackgroundTasks que facilita la ejecución de funciones después de que se haya enviado la respuesta al cliente. Esta funcionalidad es especialmente útil para operaciones que pueden tardar varios segundos pero que no afectan directamente al resultado que el usuario necesita recibir.

Importación y uso básico

Para utilizar tareas en background, necesitamos importar BackgroundTasks desde FastAPI y añadirlo como parámetro en nuestras funciones de ruta:

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def escribir_log(mensaje: str):
    """Función que se ejecutará en background"""
    with open("actividad.log", "a") as archivo:
        archivo.write(f"{mensaje}\n")

@app.post("/procesar-pedido/")
async def procesar_pedido(pedido_id: int, background_tasks: BackgroundTasks):
    # Procesamiento inmediato del pedido
    resultado = {"pedido_id": pedido_id, "estado": "procesado"}
    
    # Añadir tarea en background
    background_tasks.add_task(escribir_log, f"Pedido {pedido_id} procesado correctamente")
    
    return resultado

En este ejemplo, la función escribir_log se ejecutará después de que se envíe la respuesta al cliente. El usuario recibirá inmediatamente la confirmación del pedido sin esperar a que se escriba el archivo de log.

Añadir múltiples tareas

Podemos añadir varias tareas en background dentro de la misma función de ruta. Cada tarea se ejecutará de forma secuencial en el orden que fueron añadidas:

def enviar_email_confirmacion(email: str, pedido_id: int):
    """Simula el envío de un email de confirmación"""
    print(f"Enviando email a {email} para pedido {pedido_id}")
    # Aquí iría la lógica real de envío de email

def actualizar_estadisticas(pedido_id: int):
    """Actualiza las estadísticas de ventas"""
    print(f"Actualizando estadísticas para pedido {pedido_id}")
    # Aquí iría la lógica de actualización

@app.post("/crear-pedido/")
async def crear_pedido(email: str, background_tasks: BackgroundTasks):
    pedido_id = 12345  # Simulamos la creación del pedido
    
    # Añadir múltiples tareas en background
    background_tasks.add_task(enviar_email_confirmacion, email, pedido_id)
    background_tasks.add_task(actualizar_estadisticas, pedido_id)
    background_tasks.add_task(escribir_log, f"Nuevo pedido creado: {pedido_id}")
    
    return {"pedido_id": pedido_id, "mensaje": "Pedido creado exitosamente"}

Pasar argumentos a las tareas

Las tareas en background pueden recibir tanto argumentos posicionales como argumentos con nombre. Esto nos permite pasar toda la información necesaria a nuestras funciones:

def procesar_archivo(nombre_archivo: str, formato: str = "pdf", comprimir: bool = False):
    """Procesa un archivo con opciones específicas"""
    print(f"Procesando {nombre_archivo} en formato {formato}")
    if comprimir:
        print("Aplicando compresión al archivo")
    # Lógica de procesamiento del archivo

@app.post("/subir-archivo/")
async def subir_archivo(nombre: str, background_tasks: BackgroundTasks):
    # Simulamos la subida del archivo
    archivo_guardado = f"uploads/{nombre}"
    
    # Añadir tarea con argumentos posicionales y con nombre
    background_tasks.add_task(
        procesar_archivo, 
        archivo_guardado,           # Argumento posicional
        formato="png",              # Argumento con nombre
        comprimir=True              # Argumento con nombre
    )
    
    return {"archivo": nombre, "estado": "subido"}

Manejo de errores en tareas background

Es importante considerar que las tareas en background se ejecutan después de enviar la respuesta, por lo que los errores que ocurran en estas tareas no afectarán la respuesta HTTP. Sin embargo, debemos manejarlos adecuadamente:

def tarea_con_manejo_errores(usuario_id: int):
    """Tarea que maneja sus propios errores"""
    try:
        # Simulamos una operación que puede fallar
        if usuario_id < 0:
            raise ValueError("ID de usuario inválido")
        
        print(f"Procesando usuario {usuario_id}")
        # Lógica de procesamiento
        
    except Exception as e:
        # Registrar el error para debugging
        with open("errores.log", "a") as archivo:
            archivo.write(f"Error en tarea background: {str(e)}\n")

@app.post("/procesar-usuario/{usuario_id}")
async def procesar_usuario(usuario_id: int, background_tasks: BackgroundTasks):
    background_tasks.add_task(tarea_con_manejo_errores, usuario_id)
    
    return {"usuario_id": usuario_id, "estado": "en_proceso"}

Funciones asíncronas como tareas background

También podemos utilizar funciones asíncronas como tareas en background. Esto es útil cuando necesitamos realizar operaciones asíncronas como llamadas a APIs externas:

import httpx

async def notificar_servicio_externo(pedido_id: int):
    """Notifica a un servicio externo sobre el nuevo pedido"""
    async with httpx.AsyncClient() as client:
        try:
            response = await client.post(
                "https://api.externa.com/notificaciones",
                json={"pedido_id": pedido_id, "evento": "pedido_creado"}
            )
            print(f"Notificación enviada: {response.status_code}")
        except Exception as e:
            print(f"Error al notificar: {e}")

@app.post("/pedido-con-notificacion/")
async def crear_pedido_con_notificacion(pedido_id: int, background_tasks: BackgroundTasks):
    # Añadir función asíncrona como tarea background
    background_tasks.add_task(notificar_servicio_externo, pedido_id)
    
    return {"pedido_id": pedido_id, "notificacion": "programada"}

Las tareas en background de FastAPI proporcionan una solución sencilla y eficaz para operaciones que no requieren respuesta inmediata, mejorando significativamente la experiencia del usuario al reducir los tiempos de respuesta de la API.

Casos de uso típicos (emails, logs)

Las tareas en background son especialmente útiles en escenarios donde necesitamos realizar operaciones que pueden tardar tiempo pero que no afectan la respuesta inmediata al usuario. Los casos más comunes incluyen el envío de emails de confirmación, registro de actividades y procesamiento de archivos.

Envío de emails de confirmación

Uno de los casos más frecuentes es el envío de emails después de que un usuario realiza una acción importante como registrarse, realizar una compra o cambiar su contraseña. En lugar de hacer esperar al usuario mientras se envía el email, podemos confirmar la acción inmediatamente y enviar el email en background:

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel, EmailStr
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

app = FastAPI()

class UsuarioRegistro(BaseModel):
    nombre: str
    email: EmailStr
    telefono: str

def enviar_email_bienvenida(email: str, nombre: str):
    """Envía email de bienvenida al nuevo usuario"""
    try:
        # Configuración del servidor SMTP (ejemplo con Gmail)
        servidor = smtplib.SMTP('smtp.gmail.com', 587)
        servidor.starttls()
        servidor.login("tu_email@gmail.com", "tu_password")
        
        # Crear el mensaje
        mensaje = MIMEMultipart()
        mensaje['From'] = "tu_email@gmail.com"
        mensaje['To'] = email
        mensaje['Subject'] = "¡Bienvenido a nuestra plataforma!"
        
        cuerpo = f"""
        Hola {nombre},
        
        ¡Gracias por registrarte en nuestra plataforma!
        Tu cuenta ha sido creada exitosamente.
        
        Saludos,
        El equipo de desarrollo
        """
        
        mensaje.attach(MIMEText(cuerpo, 'plain'))
        servidor.send_message(mensaje)
        servidor.quit()
        
        print(f"Email de bienvenida enviado a {email}")
        
    except Exception as e:
        print(f"Error enviando email: {e}")

@app.post("/registrar-usuario/")
async def registrar_usuario(usuario: UsuarioRegistro, background_tasks: BackgroundTasks):
    # Simular guardado en base de datos
    usuario_id = 12345
    
    # Programar envío de email en background
    background_tasks.add_task(enviar_email_bienvenida, usuario.email, usuario.nombre)
    
    return {
        "mensaje": "Usuario registrado exitosamente",
        "usuario_id": usuario_id,
        "email_confirmacion": "programado"
    }

Sistema de logging y auditoría

El registro de actividades es fundamental para el monitoreo y debugging de aplicaciones. Las tareas en background nos permiten registrar eventos sin impactar el rendimiento de las operaciones principales:

from datetime import datetime
import json

def registrar_actividad(usuario_id: int, accion: str, detalles: dict = None):
    """Registra actividad del usuario en archivo de log"""
    timestamp = datetime.now().isoformat()
    
    entrada_log = {
        "timestamp": timestamp,
        "usuario_id": usuario_id,
        "accion": accion,
        "detalles": detalles or {}
    }
    
    try:
        with open("actividad_usuarios.log", "a", encoding="utf-8") as archivo:
            archivo.write(json.dumps(entrada_log, ensure_ascii=False) + "\n")
        
        print(f"Actividad registrada: {accion} para usuario {usuario_id}")
        
    except Exception as e:
        print(f"Error registrando actividad: {e}")

def registrar_error(endpoint: str, error: str, usuario_id: int = None):
    """Registra errores de la aplicación"""
    timestamp = datetime.now().isoformat()
    
    entrada_error = {
        "timestamp": timestamp,
        "tipo": "error",
        "endpoint": endpoint,
        "error": error,
        "usuario_id": usuario_id
    }
    
    try:
        with open("errores_aplicacion.log", "a", encoding="utf-8") as archivo:
            archivo.write(json.dumps(entrada_error, ensure_ascii=False) + "\n")
            
    except Exception as e:
        print(f"Error registrando error: {e}")

@app.post("/actualizar-perfil/{usuario_id}")
async def actualizar_perfil(usuario_id: int, datos: dict, background_tasks: BackgroundTasks):
    try:
        # Simular actualización del perfil
        print(f"Actualizando perfil del usuario {usuario_id}")
        
        # Registrar la actividad en background
        background_tasks.add_task(
            registrar_actividad,
            usuario_id,
            "perfil_actualizado",
            {"campos_modificados": list(datos.keys())}
        )
        
        return {"mensaje": "Perfil actualizado correctamente"}
        
    except Exception as e:
        # Registrar el error en background
        background_tasks.add_task(
            registrar_error,
            f"/actualizar-perfil/{usuario_id}",
            str(e),
            usuario_id
        )
        
        return {"error": "Error actualizando perfil"}, 500

Procesamiento de archivos y reportes

Otro caso común es el procesamiento de archivos subidos por usuarios, como imágenes que necesitan redimensionarse o documentos que requieren análisis:

from PIL import Image
import os

def procesar_imagen(ruta_archivo: str, usuario_id: int):
    """Procesa imagen subida: redimensiona y crea thumbnails"""
    try:
        # Abrir la imagen original
        with Image.open(ruta_archivo) as img:
            # Crear versión redimensionada (máximo 1920x1080)
            img.thumbnail((1920, 1080), Image.Resampling.LANCZOS)
            
            # Guardar versión optimizada
            nombre_base = os.path.splitext(ruta_archivo)[0]
            ruta_optimizada = f"{nombre_base}_optimizada.jpg"
            img.save(ruta_optimizada, "JPEG", quality=85, optimize=True)
            
            # Crear thumbnail (200x200)
            img_thumb = img.copy()
            img_thumb.thumbnail((200, 200), Image.Resampling.LANCZOS)
            ruta_thumbnail = f"{nombre_base}_thumb.jpg"
            img_thumb.save(ruta_thumbnail, "JPEG", quality=80)
            
        print(f"Imagen procesada para usuario {usuario_id}: {ruta_archivo}")
        
        # Registrar el procesamiento
        registrar_actividad(
            usuario_id,
            "imagen_procesada",
            {
                "archivo_original": ruta_archivo,
                "archivo_optimizado": ruta_optimizada,
                "thumbnail": ruta_thumbnail
            }
        )
        
    except Exception as e:
        print(f"Error procesando imagen: {e}")
        registrar_error("procesamiento_imagen", str(e), usuario_id)

def generar_reporte_mensual(usuario_id: int, mes: int, año: int):
    """Genera reporte mensual de actividad del usuario"""
    try:
        # Simular generación de reporte
        print(f"Generando reporte mensual para usuario {usuario_id}: {mes}/{año}")
        
        # Aquí iría la lógica real de generación del reporte
        # Por ejemplo: consultar base de datos, procesar datos, crear PDF
        
        nombre_reporte = f"reporte_{usuario_id}_{año}_{mes:02d}.pdf"
        
        # Simular creación del archivo
        with open(f"reportes/{nombre_reporte}", "w") as archivo:
            archivo.write(f"Reporte mensual para usuario {usuario_id}\n")
            archivo.write(f"Período: {mes}/{año}\n")
            archivo.write("Datos del reporte...\n")
        
        print(f"Reporte generado: {nombre_reporte}")
        
        # Registrar la generación del reporte
        registrar_actividad(
            usuario_id,
            "reporte_generado",
            {"archivo": nombre_reporte, "periodo": f"{mes}/{año}"}
        )
        
    except Exception as e:
        print(f"Error generando reporte: {e}")
        registrar_error("generacion_reporte", str(e), usuario_id)

@app.post("/subir-imagen/{usuario_id}")
async def subir_imagen(usuario_id: int, background_tasks: BackgroundTasks):
    # Simular guardado de archivo
    ruta_archivo = f"uploads/imagen_{usuario_id}_123.jpg"
    
    # Procesar imagen en background
    background_tasks.add_task(procesar_imagen, ruta_archivo, usuario_id)
    
    return {
        "mensaje": "Imagen subida correctamente",
        "procesamiento": "en_curso"
    }

@app.post("/solicitar-reporte/{usuario_id}")
async def solicitar_reporte(usuario_id: int, mes: int, año: int, background_tasks: BackgroundTasks):
    # Generar reporte en background
    background_tasks.add_task(generar_reporte_mensual, usuario_id, mes, año)
    
    return {
        "mensaje": "Reporte solicitado",
        "estado": "generando",
        "periodo": f"{mes}/{año}"
    }

Combinando múltiples tareas

En aplicaciones reales, es común combinar varios tipos de tareas en background para una sola operación:

@app.post("/completar-compra/")
async def completar_compra(
    usuario_id: int,
    email: str,
    productos: list,
    background_tasks: BackgroundTasks
):
    # Simular procesamiento de la compra
    compra_id = 98765
    total = sum(producto.get("precio", 0) for producto in productos)
    
    # Programar múltiples tareas en background
    background_tasks.add_task(
        enviar_email_bienvenida,  # Reutilizamos la función de email
        email,
        "Cliente"
    )
    
    background_tasks.add_task(
        registrar_actividad,
        usuario_id,
        "compra_completada",
        {
            "compra_id": compra_id,
            "total": total,
            "productos": len(productos)
        }
    )
    
    background_tasks.add_task(
        generar_reporte_mensual,
        usuario_id,
        datetime.now().month,
        datetime.now().year
    )
    
    return {
        "compra_id": compra_id,
        "total": total,
        "mensaje": "Compra procesada exitosamente",
        "confirmacion_email": "enviando"
    }

Estos casos de uso demuestran cómo las tareas en background mejoran significativamente la experiencia del usuario al permitir que las operaciones principales respondan rápidamente, mientras que las tareas secundarias se ejecutan de forma transparente en segundo plano.