El módulo sched implementa un scheduler simple de eventos en memoria. Mantiene una cola ordenada por instante y prioridad, espera hasta el momento adecuado y ejecuta callbacks. Es útil en scripts, simulaciones, tests, pequeños servicios y aplicaciones que necesitan temporización local sin una dependencia externa.
sched no es cron, una cola distribuida ni infraestructura durable. Los eventos desaparecen cuando termina el proceso, los callbacks largos retrasan los siguientes y el módulo no ofrece retries, persistencia ni ejecución en varias máquinas. Úsalo para scheduling local y controlado.
Crea un scheduler
El constructor recibe una función de tiempo y una función de espera. Los defaults modernos usan time.monotonic y time.sleep.
import sched
import time
scheduler = sched.scheduler(time.monotonic, time.sleep)
Un reloj monotónico es apropiado para intervalos porque no retrocede cuando cambia el reloj civil.
Programa con retraso relativo
enter() agenda una acción después de un delay.
def ejecutar(nombre):
print("ejecutando", nombre)
scheduler.enter(5, 1, ejecutar, argument=("tarea",))
scheduler.run()
El segundo argumento es la prioridad. Los números menores se ejecutan primero cuando los eventos comparten instante.
Argumentos y kwargs
Usa argument para parámetros posicionales y kwargs para los nombrados.
scheduler.enter(
2,
1,
enviar,
argument=(destino,),
kwargs={"intento": 1},
)
Los argumentos explícitos son más fáciles de probar y registrar que closures con estado mutable.
Programa en un instante absoluto
enterabs() recibe un valor compatible con la función de tiempo.
cuando = time.monotonic() + 10
scheduler.enterabs(cuando, 1, ejecutar, argument=("absoluta",))
No mezcles timestamps de time.time() con un scheduler basado en time.monotonic().
Prioridades
Cuando varios eventos tienen el mismo instante, la prioridad define el orden.
scheduler.enter(1, 10, ejecutar, argument=("normal",))
scheduler.enter(1, 1, ejecutar, argument=("urgente",))
La prioridad no interrumpe un callback que ya está ejecutándose.
Objeto Event
enter() y enterabs() devuelven un evento que puede cancelarse.
evento = scheduler.enter(30, 1, ejecutar)
Guarda la referencia junto al identificador lógico de la tarea.
Cancela un evento
cancel() elimina un evento pendiente.
try:
scheduler.cancel(evento)
except ValueError:
print("el evento ya no está pendiente")
El cancelamiento falla si ya se ejecutó, fue eliminado o no pertenece a esa cola.
Inspecciona la cola
La propiedad queue muestra eventos pendientes en orden.
for evento in scheduler.queue:
print(evento.time, evento.priority, evento.action)
Úsala para observabilidad, no para modificar directamente la estructura.
Ejecuta sin bloquear hasta el próximo evento
run(blocking=False) procesa eventos vencidos y devuelve información sobre el siguiente plazo.
proximo = scheduler.run(blocking=False)
Este modo ayuda a integrar el scheduler con una GUI o loop principal.
Callbacks largos
El scheduler ejecuta callbacks de forma secuencial. Una acción de diez segundos retrasa todos los eventos que venzan durante ese tiempo.
Mantén callbacks cortos o envía trabajo a un executor limitado. No crees una thread ilimitada por evento.
Eventos atrasados
Cuando el proceso está ocupado, sched no descarta automáticamente eventos vencidos. Los ejecuta tan pronto como puede.
La aplicación debe decidir si la tarea todavía tiene sentido. Compara el instante planificado con el real y define tolerancia.
Tareas recurrentes
Un callback puede agendar su próxima ejecución.
INTERVALO = 60
def recurrente(proximo_instante):
ejecutar_trabajo()
siguiente = proximo_instante + INTERVALO
scheduler.enterabs(siguiente, 1, recurrente, argument=(siguiente,))
primero = time.monotonic() + INTERVALO
scheduler.enterabs(primero, 1, recurrente, argument=(primero,))
Calcular desde el instante planificado reduce drift.
Evita drift
Si el callback usa enter(INTERVALO, ...) después de terminar, la duración de la tarea se suma a cada ciclo.
Usa plazos absolutos para cadencia fija y delays relativos solo cuando la regla sea esperar después de concluir.
Fallos en callbacks
Una excepción sale de run(). El scheduler mantiene su estructura, pero los eventos posteriores esperan hasta que el caller vuelva a ejecutar run().
def seguro():
try:
ejecutar_trabajo()
except Exception:
logger.exception("evento falló")
Captura errores solo cuando exista una política clara.
Retries
sched no repite automáticamente.
Reintenta únicamente fallos transitorios, con límite, backoff, jitter e idempotencia. No repitas errores de validación.
Threads
Las implementaciones modernas permiten insertar eventos desde varias threads, pero el lifecycle y los callbacks todavía requieren diseño.
No mantengas locks de aplicación mientras ejecutas callbacks que puedan necesitar los mismos recursos.
Eventos anteriores insertados por otra thread
Si una thread espera un evento lejano y otra inserta uno más cercano, la integración externa debe hacer que el loop reevalúe la cola.
Para sistemas complejos, puede ser mejor asyncio o un scheduler dedicado.
Integra ThreadPoolExecutor
El scheduler puede enviar trabajo a un pool limitado.
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor(max_workers=4)
scheduler.enter(1, 1, executor.submit, argument=(ejecutar_trabajo,))
Observa futures y cierra el executor durante shutdown. Consulta concurrent.futures en Python.
Shutdown
Mantén una flag o evento de parada. Deja de crear recurrencias, cancela pendientes cuando corresponda y espera trabajo enviado.
Un scheduler en memoria no conserva tareas tras el cierre.
Horario civil
Para reglas como “09:00 local”, usa una solución que gestione timezone, horario de verano y persistencia.
time.monotonic() mide intervalos; no representa una fecha de calendario.
Tests determinísticos
Inyecta funciones falsas de tiempo y delay.
class Reloj:
def __init__(self):
self.ahora = 0
def time(self):
return self.ahora
def sleep(self, retraso):
self.ahora += retraso
reloj = Reloj()
test_scheduler = sched.scheduler(reloj.time, reloj.sleep)
Los tests avanzan sin sleeps reales.
Persistencia
Si los eventos deben sobrevivir a restarts, guarda la intención en base o cola durable y reconstruye al iniciar.
No serialices callbacks arbitrarios. Persiste un tipo aprobado y parámetros validados.
Seguridad
No permitas que usuarios agenden callables Python arbitrarios. Mapea comandos permitidos a funciones conocidas.
Aplica límites de cantidad, frecuencia, prioridad y tamaño.
Observabilidad
Registra ID lógico, instante planificado, inicio real, retraso, duración, resultado e intento.
Métricas de cola, lateness y fallos muestran callbacks bloqueantes.
Cuándo usar otra herramienta
Usa cron o Task Scheduler para procesos independientes, asyncio para apps asíncronas, colas durables para trabajo persistente y bibliotecas especializadas para calendarios complejos.
sched encaja en una cola temporal local y pequeña.
Errores comunes
Los fallos frecuentes son mezclar relojes, ejecutar callbacks largos, introducir drift, asumir persistencia, ignorar excepciones, olvidar cancelación, tratar prioridad como preemption y usar sleeps reales en tests.
Conclusión
sched ofrece scheduling local mediante una cola temporal. Usa reloj monotónico para intervalos, deadlines absolutos para cadencia fija, prioridades para desempate y callbacks cortos.
Añade cancelación, observabilidad y una política explícita de errores. Consulta la documentación oficial de sched y concurrent.futures en Python.
