La integración de un sistema de tareas programdaas es un requisito común en aplicaciones web. Dentro del ecosistema Python, APScheduler se destaca como una solución robusta y flexilbe, especialmente adecuada para aplicaciones basadas en frameworks como FastAPI. A diferencia de soluciones como Celery, que requieren un broker de mensajes (como Redis o RabbitMQ) para un procesamiento distribuido y asíncrono, APScheduler puede funcionar autónomamante con almacenamiento local o en base de datos.

Un error común al integrar APScheduler en una aplicación web es utilizar el BlockingScheduler, ya que este detiene la ejecución del hilo principal y bloquea el servidor web. Para entornos como FastAPI o Flask, se debe emplear el BackgroundScheduler.

Configuración fundamental del planificador

El siguiente fragmento muestra una configuración base que define almacenes de trabajos, ejecutores y el propio planificador. Se utiliza una base de datos SQLite para persistir los trabajos, lo que permite que sobrevivan a reinicios de la aplicación.

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ThreadPoolExecutor
from pytz import utc

# Definición de almacén de trabajos y ejecutores
almacen_trabajos = {'persistente': SQLAlchemyJobStore(url='sqlite:///mis_trabajos.db')}
ejecutores = {'defecto': ThreadPoolExecutor(max_workers=20)}

# Instanciación del planificador
planificador = BackgroundScheduler(
    jobstores=almacen_trabajos,
    executors=ejecutores,
    timezone=utc
)

# Ejemplo de una tarea simple
def mostrar_mensaje(mensaje: str):
    print(f"Tarea ejecutada: {mensaje}")

Endpoints de gestión con FastAPI

Para controlar el planificador durante la ejecución, se crean rutas en la aplicación FastAPI. Estos endpoints permiten listar, pausar, reanudar y eliminar tareas de manera dinámica.

from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse

app = FastAPI()

@app.get("/tareas", summary="Lista todas las tareas programadas")
async def listar_tareas():
    trabajos = planificador.get_jobs()
    resultado = []
    for trabajo in trabajos:
        info = {
            "id": trabajo.id,
            "nombre": trabajo.name,
            "proxima_ejecucion": str(trabajo.next_run_time),
            "estado": "Activo" if trabajo.next_run_time else "Pausado"
        }
        resultado.append(info)
    return JSONResponse(content=resultado)

@app.post("/tareas/{trabajo_id}/pausar", summary="Pausa una tarea específica")
async def pausar_tarea(trabajo_id: str):
    trabajo = planificador.get_job(trabajo_id)
    if not trabajo:
        raise HTTPException(status_code=404, detail="Tarea no encontrada")
    trabajo.pause()
    return {"mensaje": f"Tarea {trabajo_id} pausada"}

@app.post("/tareas/{trabajo_id}/reanudar", summary="Reanuda una tarea pausada")
async def reanudar_tarea(trabajo_id: str):
    trabajo = planificador.get_job(trabajo_id)
    if not trabajo:
        raise HTTPException(status_code=404, detail="Tarea no encontrada")
    trabajo.resume()
    return {"mensaje": f"Tarea {trabajo_id} reanudada"}

@app.delete("/tareas/{trabajo_id}", summary="Elimina una tarea definitivamente")
async def eliminar_tarea(trabajo_id: str):
    trabajo = planificador.get_job(trabajo_id)
    if not trabajo:
        raise HTTPException(status_code=404, detail="Tarea no encontrada")
    trabajo.remove()
    return {"mensaje": f"Tarea {trabajo_id} eliminada"}

Integración con el ciclo de vida de FastAPI

Es esencial iniciar y detener el planificador de manera sincronizada con el ciclo de vida de la aplicación FastAPI. Esto se logra utilizando los eventos startup y shutdown.

@app.on_event("startup")
async def iniciar_planificador():
    # Aquí se pueden añadir trabajos iniciales
    planificador.add_job(
        mostrar_mensaje,
        'interval',
        args=["Tarea periódica cada 60 segundos"],
        seconds=60,
        id="tarea_periodica_1"
    )
    planificador.start()
    print("Planificador APScheduler iniciado.")

@app.on_event("shutdown")
async def detener_planificador():
    planificador.shutdown()
    print("Planificador APScheduler detenido.")