sched en Python: programa eventos

No toda tarea programada necesita un servicio externo, una cola distribuida o cron del sistema. Para scripts, simulaciones, tests y procesos locales que deben ejecutar funciones en momentos definidos, el módulo sched en Python ofrece un scheduler pequeño dentro de la biblioteca estándar. Ordena eventos por instante y prioridad, espera hasta el momento adecuado y llama funciones con los argumentos definidos por la aplicación.

Esta guía explica scheduler, enter(), enterabs(), cancelación, ejecución bloqueante y no bloqueante y tareas recurrentes. Complementa nuestros artículos sobre atexit en Python, tracebacks, contextvars, collections y scripts Python lentos.

Qué resuelve sched

Un scheduler recibe eventos y decide cuándo ejecutar cada acción. Todo en sched vive dentro del proceso actual. No existe persistencia automática, base de datos, worker remoto ni recuperación después de reiniciar.

El módulo resulta apropiado para:

• llamar un callback después de un retraso;
• simular eventos temporales en tests;
• programar trabajos cortos en una herramienta local;
• organizar pasos temporizados de una demostración;
• crear un loop periódico ligero.

Las tareas que deben sobrevivir a crashes, distribuir carga o ejecutarse en varias máquinas necesitan infraestructura persistente.

Crear un scheduler

La clase principal es sched.scheduler.

import sched

programador = sched.scheduler()

Las versiones actuales usan time.monotonic como reloj y time.sleep para esperar. Un reloj monotónico evita que ajustes del reloj civil muevan plazos relativos.

La documentación oficial de sched indica que la clase puede utilizarse con seguridad en entornos multithread desde Python 3.3, aunque los datos compartidos por los callbacks todavía necesitan sincronización propia.

Programar después de un retraso

enter() recibe retraso, prioridad, acción, secuencia de argumentos y argumentos nombrados opcionales.

import sched

programador = sched.scheduler()


def avisar(mensaje):
    print(mensaje)

programador.enter(
    delay=2,
    priority=1,
    action=avisar,
    argument=("Pasaron dos segundos",),
)

programador.run()

Con el reloj y la espera predeterminados, el retraso se expresa en segundos. El método devuelve un objeto de evento que puede cancelarse.

Argumentos posicionales y nombrados

argument debe ser una secuencia, normalmente una tupla. kwargs recibe un diccionario.

def registrar(nombre, estado="ok"):
    print(nombre, estado)

programador.enter(
    1,
    1,
    registrar,
    argument=("importación",),
    kwargs={"estado": "completa"},
)

El evento guarda referencias. Si un argumento mutable cambia antes de ejecutar, el callback verá el objeto modificado.

Programar en un instante absoluto

enterabs() recibe un valor absoluto producido por la función de tiempo configurada.

import time

programador = sched.scheduler(
    timefunc=time.monotonic,
    delayfunc=time.sleep,
)

instante = time.monotonic() + 5
programador.enterabs(
    instante,
    1,
    avisar,
    argument=("Llegó el plazo monotónico",),
)

No pases un timestamp de time.time() a un scheduler basado en time.monotonic(). Ambos relojes tienen orígenes y finalidades diferentes.

Tiempo monotónico frente a hora civil

time.monotonic() solo avanza y no se ve afectado por cambios manuales, correcciones NTP ni horario de verano. Es la mejor opción para retrasos e intervalos.

time.time() representa una escala próxima al tiempo Unix y puede ser necesario para “ejecutar a las 15:00 de esta fecha”. Convierte valores de calendario con cuidado y considera cambios del reloj.

La documentación oficial de time explica los relojes disponibles. La programación por calendario y zonas horarias suele requerir datetime y zoneinfo antes de llegar a sched.

Prioridad de los eventos

Cuando dos eventos tienen el mismo instante, los números de prioridad menores se ejecutan primero.

programador.enter(1, 20, avisar, argument=("prioridad 20",))
programador.enter(1, 5, avisar, argument=("prioridad 5",))
programador.enter(1, 10, avisar, argument=("prioridad 10",))

La prioridad no interrumpe una acción ya iniciada. Solo organiza los eventos pendientes.

Ejecutar la cola

run() procesa eventos en orden.

programador.run(blocking=True)

En modo bloqueante, el método espera hasta ejecutar todos los eventos que estaban en la cola. La función de retraso recibe el tiempo restante antes de cada evento.

Los eventos atrasados no se descartan

Si una acción tarda más que el intervalo hasta el siguiente evento, el scheduler queda atrasado. Los eventos vencidos se ejecutan tan pronto como sea posible y ninguno se elimina.

import time


def tarea_lenta():
    time.sleep(3)
    print("terminó la tarea lenta")

programador.enter(0, 1, tarea_lenta)
programador.enter(1, 1, avisar, argument=("evento atrasado",))
programador.run()

Si el retraso acumulado no es aceptable, envía trabajos largos a workers controlados o limita el callback a despachar trabajo.

Ejecución no bloqueante

Con blocking=False, run() ejecuta los eventos vencidos y devuelve el próximo plazo o None cuando la cola está vacía.

proximo = programador.run(blocking=False)

if proximo is None:
    print("Cola vacía")
else:
    print("Próximo plazo:", proximo)

Este modo permite integrarlo con otro loop. Interpreta el valor retornado de acuerdo con la versión y el reloj configurado.

Integrar con un loop propio

import time

while not programador.empty():
    plazo = programador.run(blocking=False)
    if plazo is None:
        break

    ahora = time.monotonic()
    retraso = max(0, plazo - ahora)
    procesar_interfaz_durante(min(retraso, 0.1))

Los frameworks gráficos o asíncronos suelen tener timers propios. Programa el próximo despertar con el framework en lugar de dormir la thread principal.

Inspeccionar la cola

La propiedad queue devuelve los eventos pendientes en orden de ejecución.

for evento in programador.queue:
    print(
        evento.time,
        evento.priority,
        evento.action,
        evento.argument,
        evento.kwargs,
    )

La lista es un snapshot ordenado. Modificarla no cambia la cola interna.

Comprobar si está vacía

empty() informa si hay eventos pendientes.

if programador.empty():
    print("No hay eventos")

En una aplicación multithread, otra thread puede cambiar el estado inmediatamente después.

Cancelar un evento

Conserva el objeto devuelto por enter() o enterabs() y pásalo a cancel().

evento = programador.enter(
    30,
    1,
    avisar,
    argument=("no se ejecutará",),
)

programador.cancel(evento)

Si el evento ya no está en la cola, cancel() lanza ValueError.

Cancelación concurrente

try:
    programador.cancel(evento)
except ValueError:
    print("El evento ya se ejecutó o fue cancelado")

Otra thread puede iniciar el callback entre la decisión y la cancelación. Una tarea que necesita garantía fuerte debe consultar una flag propia antes de producir efectos.

Tareas recurrentes

Sched no posee un tipo especial de trabajo recurrente. El callback programa la siguiente ejecución.

def ejecutar_periodicamente(intervalo):
    print("ejecutando")
    programador.enter(
        intervalo,
        1,
        ejecutar_periodicamente,
        argument=(intervalo,),
    )

programador.enter(0, 1, ejecutar_periodicamente, argument=(5,))

Este modelo mide el intervalo desde el momento de reprogramación. Si el callback tarda dos segundos, el intervalo entre inicios puede ser de siete.

Evitar drift

Para mantener una cuadrícula absoluta, calcula el siguiente plazo desde el anterior.

def periodica(plazo, intervalo):
    print("ejecución periódica")
    siguiente = plazo + intervalo
    programador.enterabs(
        siguiente,
        1,
        periodica,
        argument=(siguiente, intervalo),
    )

inicio = time.monotonic() + 1
programador.enterabs(inicio, 1, periodica, argument=(inicio, 5))

Define qué ocurre si se pierden varias ejecuciones: ejecutar todas, saltar algunas o calcular el próximo instante futuro.

Excepciones en callbacks

Si una acción genera una excepción, run() la propaga. El estado interno permanece consistente y el evento fallido no se reintenta automáticamente.

def protegido():
    try:
        operacion()
    except Exception:
        logger.exception("Falló el evento programado")

programador.enter(1, 1, protegido)

Elige propagar o capturar según la responsabilidad de la aplicación. Las fallas críticas deben ser observables.

Uso con threads

La cola es thread-safe, por lo que una thread puede añadir eventos mientras otra ejecuta run(). Los callbacks y datos compartidos no reciben protección automática.

from threading import Lock

lock = Lock()
estado = {}


def actualizar(clave, valor):
    with lock:
        estado[clave] = valor

Un run(blocking=True) que ya está esperando puede no reaccionar inmediatamente a un evento más temprano añadido desde otra thread. Los sistemas responsivos usan comprobaciones cortas o un mecanismo de despertar propio.

Funciones de tiempo personalizadas

Las funciones personalizadas permiten tests determinísticos.

class RelojFalso:
    def __init__(self):
        self.ahora = 0.0

    def tiempo(self):
        return self.ahora

    def esperar(self, segundos):
        self.ahora += segundos

reloj = RelojFalso()
programador = sched.scheduler(reloj.tiempo, reloj.esperar)

Los tests avanzan tiempo virtual sin esperar segundos reales.

delayfunc recibe cero

Después de cada evento, sched llama la función de espera con cero para dar oportunidad a otras threads.

Una función personalizada debe aceptar cero y no tratarlo como error.

Tests determinísticos

eventos = []

programador.enter(5, 1, eventos.append, argument=("A",))
programador.enter(2, 1, eventos.append, argument=("B",))
programador.run()

assert eventos == ["B", "A"]
assert reloj.ahora == 5

Prueba además misma hora con prioridades diferentes, cancelación, excepciones y eventos añadidos durante la ejecución.

Persistencia y reinicio

La cola solo existe en memoria. Todos los eventos desaparecen al terminar el proceso.

Los trabajos que deben sobrevivir reinicios deben guardarse en una base o cola durable y reconstruirse durante el startup. Evita serializar callables arbitrarios con pickle.

Varias máquinas

Sched no coordina locks distribuidos ni impide ejecución duplicada en varias instancias. Cada proceso tiene una cola independiente.

Usa un scheduler distribuido, leasing en base de datos o una cola de tareas cuando solo una instancia debe ejecutar.

Observabilidad

Registra al menos:

• identificador lógico del evento;
• hora prevista y real;
• retraso acumulado;
• duración;
• estado y excepción;
• cantidad de eventos pendientes.

No registres argumentos con contraseñas, tokens o datos personales.

Cierre controlado

Impide nuevas reprogramaciones, cancela eventos pendientes cuando corresponde y permite terminar callbacks activos.

cerrando = False


def recurrente(intervalo):
    ejecutar()
    if not cerrando:
        programador.enter(intervalo, 1, recurrente, argument=(intervalo,))

Coordina este protocolo con señales y el ciclo de vida de la aplicación. Atexit puede registrar una métrica final, pero no garantiza ejecución durante un crash.

Errores frecuentes

• Mezclar time.time() y time.monotonic().
• Ejecutar callbacks largos en la thread del scheduler.
• Suponer que los eventos atrasados se descartan.
• Cancelar sin manejar ValueError.
• Crear drift recurrente sin intención.
• Usar una cola en memoria para trabajos críticos.
• Compartir estado sin locks.
• Tratar sched como scheduler distribuido.

Buenas prácticas

• Usa reloj monotónico para intervalos.
• Mantén callbacks cortos y observables.
• Conserva objetos de evento para cancelación.
• Define políticas para retrasos y recurrencia.
• Usa reloj falso en tests.
• Sincroniza datos compartidos.
• Persiste trabajos que deben sobrevivir.
• Elige otro sistema para varias máquinas.

Conclusión

El módulo sched en Python proporciona una cola temporal sencilla para ejecutar funciones después de retrasos o en plazos absolutos. Ordena eventos por instante y prioridad, soporta cancelación, permite inspeccionar la cola y funciona con relojes personalizados.

Su fortaleza es la simplicidad local. Con callbacks cortos, el reloj correcto y políticas explícitas de retraso, recurrencia y cierre, sched resuelve timers internos sin dependencias. La persistencia, distribución y recuperación requieren otro sistema.