¿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.
CPU
1 vCPU
MEMORIA
1 GB
ALMACENAMIENTO
10 GB
TRANSFERENCIA
1 TB
PRECIO
$ 4 mes
Para obtener el servidor GRATIS debes de escribir el cupon "LEIFER"

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:

Fuente