Saltar a contenido

0015. Tareas programadas (APScheduler)

Estado: Propuesto Fecha: 2026-09-04 Autor: Duval Alcivar Módulo(s) afectado(s): todos (core, ERP y CRM) — transversal

Contexto

Hay tareas que deben ejecutarse de forma periódica o a una hora exacta en cada microservicio: crear/caer particiones de tablas históricas (audit_log, outbox, stock), purgar tokens expirados, despachar el outbox de eventos, refrescar cachés, programar reportes o sincronizar con integraciones externas.

No hay que depender de cron del sistema operativo, de CronJob de Kubernetes ni de extensiones de PostgreSQL: este ADR fija APScheduler como el "cron general" en Python, el equivalente al Scheduler de Laravel.

Decisión

Cada microservicio que necesite tareas programadas usa APScheduler con un BackgroundScheduler que arranca dentro del proceso del servicio. Los jobs son idempotentes (se pueden reejecutar sin daño) y viven versionados en el repo del servicio, en app/jobs/.

Por qué APScheduler y no pg_partman/pg_cron: APScheduler es puro Python, portable a cualquier PostgreSQL (incluido managed) y deja la lógica versionada en el repo. pg_partman/pg_cron exigen instalar extensiones en el servidor (no disponible en todos los RDS/Aurora "managed").

1. Instalar la dependencia

# pyproject.toml  ([tool.poetry.dependencies])
apscheduler = "^3.10.0"

poetry add apscheduler        # agrega la dependencia y fija poetry.lock

2. Job de ejemplo: particionado de audit_log

El caso canónico es el particionado mensual de audit_log (ADR 0014): crear las particiones del mes actual y de los +2 siguientes, y descartar las que superan la retención.

# app/jobs/partitions.py
from datetime import datetime

from sqlalchemy import text
from sqlalchemy.orm import Session

RETENCION_MESES = 12        # cuántos meses conservar antes de descartar
PREMADE_MESES = 2           # particiones creadas por adelantado hacia el futuro


def _mes_absoluto(fecha: datetime) -> int:
    return fecha.year * 12 + (fecha.month - 1)


def _desde_mes_absoluto(total_meses: int) -> datetime:
    y, m = divmod(total_meses, 12)
    return datetime(y, m + 1, 1)


def crear_particiones_audit_log(session: Session) -> list[str]:
    """Crea las particiones faltantes (mes actual + PREMADE_MESES) y retorna las creadas.

    Idempotente: consulta pg_class antes de crear (CREATE TABLE IF NOT EXISTS
    ... PARTITION OF no es válido en todas las versiones de Postgres).
    """
    hoy = datetime.now().replace(day=1)          # usar tz-aware: ZoneInfo("America/Guayaquil")
    creadas: list[str] = []

    base = _mes_absoluto(hoy)
    for offset in range(0, PREMADE_MESES + 1):
        mes = _desde_mes_absoluto(base + offset)
        mes_siguiente = _desde_mes_absoluto(base + offset + 1)
        nombre = f"audit_log_{mes.strftime('%Y_%m')}"

        existe = session.execute(
            text("SELECT 1 FROM pg_class WHERE relname = :n"), {"n": nombre}
        ).scalar()

        if not existe:
            session.execute(
                text(
                    "CREATE TABLE erp.{n} PARTITION OF erp.audit_log "
                    "FOR VALUES FROM (:desde) TO (:hasta)"
                ),
                {"n": nombre, "desde": mes, "hasta": mes_siguiente},
            )
            session.commit()
            creadas.append(nombre)

    return creadas

# Retención (política por producto): DETACH + DROP de meses viejos
def caer_particiones_vencidas(session: Session, retencion_meses: int) -> None:
    hoy = datetime.now().replace(day=1)
    lim = _mes_absoluto(hoy) - retencion_meses
    limite = _desde_mes_absoluto(lim)
    viejo = f"audit_log_{limite.strftime('%Y_%m')}"
    session.execute(text(f"ALTER TABLE erp.audit_log DETACH PARTITION erp.{viejo}"))
    session.execute(text(f"DROP TABLE IF EXISTS erp.{viejo}"))
    session.commit()

3. Programar el job al arrancar el servicio

# app/main.py
import atexit

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger

from app.jobs.partitions import crear_particiones_audit_log

scheduler = BackgroundScheduler(timezone="America/Guayaquil")

# Corre el día 1 de cada mes a las 03:00 (mes actual + 2 de margen)
scheduler.add_job(
    crear_particiones_audit_log,
    CronTrigger(day=1, hour=3, minute=0),
    args=[get_db_session],          # tu factory de sesión inyectable
    id="crear-particiones-audit-log",
    replace_existing=True,
)

scheduler.start()
atexit.register(lambda: scheduler.shutdown())

Cron de APScheduler: CronTrigger(day=1, hour=3, minute=0) equivale a 0 3 1 * * en crontab y a ->monthlyOn(1)->at('03:00') en Laravel. Con PREMADE_MESES=2, siempre existen las particiones del mes actual y las 2 siguientes.

4. ¿Toca algo a nivel de servidor?

No. Con APScheduler no se instala nada en PostgreSQL ni en el sistema operativo: - No requiere extensiones (pg_partman, pg_cron). - No requiere crontab del SO ni CronJob de Kubernetes: el job vive dentro del proceso del microservicio y corre solo.

Lo único que debe existir es que el servicio esté corriendo (el scheduler arranca con él). Con múltiples réplicas dos pods podrían ejecutar el job a la vez; como el job es idempotente (consulta pg_class antes de crear) no hay daño, pero si se quiere evitar trabajo duplicado se deja el scheduler solo en la réplica líder o se usa pg_advisory_lock alrededor del job.

5. Otro ejemplo: sirve para mucho más que particionar

APScheduler es el "cron general" de cada servicio. Mismo patrón, otras tareas:

# app/jobs/catalogos.py — refresco del catálogo de empresas en caché
def refrescar_cache_catalogos(session: Session):
    # p.ej. re-leer mae_empresas y refrescar caché
    ...

scheduler.add_job(refrescar_cache_catalogos, CronTrigger(hour="*", minute=0), id="refresh-cat-cada-hora")

# app/jobs/outbox.py — despacho de eventos pendientes (outbox, ADR 0003)
def despachar_outbox(session: Session):
    ...  # leer registros outbox sin publicar y enviarlos

scheduler.add_job(despachar_outbox, CronTrigger(minute="*/1"), id="outbox-cada-minuto")

# app/jobs/notificaciones.py — recordatorio puntual (una sola vez)
from apscheduler.triggers.date import DateTrigger
def enviar_recordatorio():
    ...
scheduler.add_job(enviar_recordatorio, DateTrigger(run_date="2026-09-15 09:00:00"), id="recordatorio-unico")

Usos típicos en los 14 servicios: purga de tokens expirados, despacho de outbox, generación de reportes programados, sincronización con integraciones externas, y particionado de históricos (audit_log, outbox, stock).

Alternativas descartadas

  • pg_partman/pg_cron (extensión de Postgres) — descartado: exige instalar extensiones en el servidor (no disponible en todos los RDS/Aurora "managed") y saca la lógica del repo del servicio.
  • cron del sistema operativo / CronJob de Kubernetes — descartado: requiere infraestructura adicional por servicio y no queda versionado junto al código ni a las migraciones.
  • Celery Beat — descartado por ahora: añade un broker (Redis/RabbitMQ) que solo compensa si ya hay un clúster de workers; APScheduler cubre el caso sin dependencia extra.

Consecuencias

  • Ganancia: tareas programadas dentro de cada servicio, versionadas en el repo, sin dependencias de infraestructura ni de extensiones del motor.
  • Ganancia: un solo patrón (APScheduler) para todo tipo de tarea periódica o puntual, consistente entre los 14 servicios.
  • Costo: el scheduler es por-proceso; con réplicas hay que cuidar la idempotencia (o dejar el scheduler en la réplica líder).
  • Costo: hay que monitorear el job (si el particionado falla meses seguidos, los INSERT sin partición fallan → alerta).
  • Pendiente: política de retención por tabla/producto, y zona horaria corporativa (America/Guayaquil) como configuración, no hardcodeada en cada job.