Implementación de la programación de trabajos en segundo plano en FastAPI con APScheduler

¿Por qué es necesaria la programación?

La programación de tareas es crucial para optimizar la utilización de los recursos, gestionar eficazmente las cargas de trabajo y automatizar las tareas rutinarias.

Basta ya de teoría, pasemos a la aplicación real

1. Cómo empezar

Antes de sumergirnos en el código, asegurémonos de tener instaladas las herramientas necesarias. Necesitarás Python junto con las librerías FastAPI y APScheduler. Instalarlas es tan fácil como usar pip:

pip install fastapi
pip install uvicorn
pip install fastapi apscheduler

2. Dependencia

Comenzamos importando los módulos necesarios y creando una instancia de FastAPI:

from fastapi import FastAPI
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.jobstores.memory import MemoryJobStore

# Create a FastAPI app instance
app = FastAPI()

3. Configuración

Configurar el almacén de trabajos y el programador A continuación, configuramos el almacén de trabajos e inicializamos una instancia de AsyncIOScheduler:

jobstores = {
    'default': MemoryJobStore()
}

# Initialize an AsyncIOScheduler with the jobstore
scheduler = AsyncIOScheduler(jobstores=jobstores, timezone='Asia/Kolkata')

Aquí estamos almacenando nuestro ID de trabajo en la Memoria no en el sistema de almacenamiento de persistencia.
Si quieres almacenar datos en un almacén de persistencia puedes almacenarlos.

4. Definiendo el Planificador

Definiendo Trabajos Programados Definimos trabajos programados de ejemplo utilizando diferentes métodos de programación:

# Job running every 10 seconds
@scheduler.scheduled_job('interval', seconds=10)
def scheduled_job_1():
    print("scheduled_job_1")

# Job running at a specific date and time
@scheduler.scheduled_job('date', run_date='2024-07-21 11:00:00')
def scheduled_job_2():
    print("scheduled_job_2")

# Job running daily at 23:44:00
@scheduler.scheduled_job('cron', day_of_week='mon-sun', hour=23, minute=44, second=0)
def scheduled_job_3():
    print("scheduled_job_3")

5. Manejo

Manejo de Eventos de la Aplicación Registramos manejadores de eventos para el inicio y cierre de la aplicación para manejar el ciclo de vida del programador:

@app.on_event("startup")
async def startup_event():
    scheduler.start()

@app.on_event("shutdown")
async def shutdown_event():
    scheduler.shutdown()

6. Endpoint

Creación de un endpoint de prueba Por último, definimos un endpoint simple ("/") para verificar la funcionalidad de la aplicación:

@app.get("/")
async def read_root():
    return {"message": "Hello, World!"}

Poniendo todo junto

Aquí está el código completo para nuestro trabajo de programación en la aplicación FastAPI:

# main.py
from fastapi import FastAPI
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.jobstores.memory import MemoryJobStore
import os

# Create a FastAPI app instance
app = FastAPI()

# Initialize a SQLAlchemyJobStore with SQLite database
jobstores = {
    'default': MemoryJobStore()
}

# Initialize an AsyncIOScheduler with the jobstore
scheduler = AsyncIOScheduler(jobstores=jobstores, timezone='Asia/Kolkata') 


# This is a scheduled job that will run every 10 seconds.
@scheduler.scheduled_job('interval', seconds=10)
def scheduled_job_1():
    print("scheduled_job_1")

@scheduler.scheduled_job('date',  run_date='2024-07-21 11:00:00' )
def scheduled_job_2():
    print("scheduled_job_2")

@scheduler.scheduled_job('cron', day_of_week='mon-sun', hour=23, minute=44, second=0)  # Decorator for scheduling the job
def scheduled_job_3():  # Function to be executed at the scheduled time
    print("scheduled_job_3")  

# Register an event for application startup
@app.on_event("startup")
async def startup_event():
    scheduler.start()

@app.on_event("shutdown")
async def shutdown_event():
    scheduler.shutdown()

@app.get("/")
async def read_root():
    return {"message": "Hello, World!"}

Para ejecutar las aplicaciones

uvicorn main:app --reload

Referencia:

https://fastapi.tiangolo.com/
https://apscheduler.readthedocs.io/en/3.x/userguide.html

Gracias por leernos. Si encuentras algo incorrecto o mejores formas de hacerlo, házmelo saber en los comentarios de abajo.

Fuente