O módulo sched implementa um agendador simples de eventos em memória. Ele mantém uma fila ordenada por instante e prioridade, espera até o momento adequado e executa callbacks. É útil em scripts, simulações, testes, pequenos serviços, tarefas temporizadas e aplicações que precisam coordenar eventos sem instalar uma biblioteca externa.
sched não é um sistema de cron, fila distribuída ou serviço persistente. Os eventos desaparecem quando o processo termina, callbacks longos atrasam os próximos e o módulo não oferece retries, armazenamento durável ou execução em múltiplas máquinas. Use-o para scheduling local e controlado.
Crie um scheduler
O construtor recebe uma função de tempo e uma função de espera. Os padrões modernos usam time.monotonic e time.sleep.
import sched
import time
scheduler = sched.scheduler(time.monotonic, time.sleep)
Um relógio monotônico é adequado para intervalos porque não volta quando o relógio civil é ajustado.
Agende com atraso relativo
enter() agenda uma ação após um atraso.
def executar(nome):
print("executando", nome)
scheduler.enter(5, 1, executar, argument=("tarefa",))
scheduler.run()
O segundo argumento é a prioridade. Números menores executam antes quando dois eventos possuem o mesmo instante.
Argumentos e keyword arguments
Use argument para argumentos posicionais e kwargs para argumentos nomeados.
scheduler.enter(
2,
1,
enviar,
argument=(destino,),
kwargs={"tentativa": 1},
)
Evite closures que capturam objetos mutáveis sem necessidade. Argumentos explícitos facilitam testes e logs.
Agende em um instante absoluto
enterabs() recebe um valor compatível com a função de tempo do scheduler.
quando = time.monotonic() + 10
scheduler.enterabs(quando, 1, executar, argument=("absoluta",))
Não misture timestamps de time.time() com um scheduler baseado em time.monotonic().
Prioridades
Quando eventos compartilham o mesmo instante, a prioridade decide a ordem.
scheduler.enter(1, 10, executar, argument=("normal",))
scheduler.enter(1, 1, executar, argument=("urgente",))
Prioridade não interrompe um callback que já começou. Ela só ordena eventos ainda pendentes.
O objeto Event
enter() e enterabs() retornam um objeto que pode ser usado para cancelamento.
evento = scheduler.enter(30, 1, executar)
Guarde a referência em uma estrutura associada ao identificador lógico da tarefa.
Cancele um evento
cancel() remove um evento pendente.
try:
scheduler.cancel(evento)
except ValueError:
print("o evento já saiu da fila")
O cancelamento falha se o evento já foi executado, removido ou não pertence à fila atual.
Inspecione a fila
A propriedade queue oferece uma lista de eventos pendentes em ordem de execução.
for evento in scheduler.queue:
print(evento.time, evento.priority, evento.action)
Use para observabilidade, não para modificar diretamente a estrutura interna.
Execute sem bloquear até o próximo evento
run(blocking=False) executa eventos vencidos e retorna o prazo do próximo, quando houver.
proximo = scheduler.run(blocking=False)
Esse modo ajuda a integrar o scheduler a um loop externo, GUI ou sistema que precisa realizar outras tarefas.
Callbacks longos
O scheduler executa callbacks de forma sequencial. Se uma ação demora dez segundos, eventos vencidos durante esse período ficam atrasados.
Mantenha callbacks curtos ou envie o trabalho a um executor controlado. Não crie uma thread ilimitada por evento.
Eventos atrasados
Quando o processo está ocupado, sched não descarta automaticamente eventos atrasados. Ele os executa assim que possível na ordem definida.
A aplicação precisa decidir se uma tarefa vencida ainda faz sentido. Compare o horário planejado com o atual e aplique uma tolerância.
Tarefas recorrentes
Um callback pode agendar a próxima execução.
INTERVALO = 60
def recorrente(proximo_instante):
executar_trabalho()
novo = proximo_instante + INTERVALO
scheduler.enterabs(novo, 1, recorrente, argument=(novo,))
inicio = time.monotonic() + INTERVALO
scheduler.enterabs(inicio, 1, recorrente, argument=(inicio,))
Calcular a próxima execução a partir do instante planejado reduz drift.
Evite drift
Se o callback usar enter(INTERVALO, ...) depois de terminar, a duração da tarefa será adicionada a cada ciclo.
Para uma cadência fixa, derive o próximo prazo da programação anterior. Para “esperar N segundos depois de concluir”, use atraso relativo conscientemente.
Falhas em callbacks
Uma exceção sai de run(). O scheduler mantém estado consistente, mas eventos seguintes não são executados até que o caller volte a chamar run().
def seguro():
try:
executar_trabalho()
except Exception:
logger.exception("evento falhou")
Capture apenas quando existir uma política clara. Falhas críticas podem precisar encerrar o processo.
Retries
sched não repete eventos automaticamente.
Implemente retry para erros transitórios com limite, backoff, jitter e idempotência. Não repita indefinidamente nem transforme erro de validação em retry.
Threads
O scheduler pode receber eventos de várias threads em versões modernas, mas o desenho da aplicação ainda precisa coordenar lifecycle, cancelamento e callbacks.
Não mantenha locks da aplicação enquanto executa callbacks, pois eles podem tentar adquirir os mesmos recursos.
Acordar após inserir evento anterior
Se uma thread está bloqueada esperando um evento distante e outra insere um evento mais próximo, a integração precisa garantir que o loop reavalie a fila.
Para sistemas concorrentes complexos, um loop baseado em Condition, asyncio ou biblioteca dedicada pode ser mais adequado.
Integração com ThreadPoolExecutor
O scheduler pode despachar trabalho para um pool limitado.
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor(max_workers=4)
scheduler.enter(1, 1, executor.submit, argument=(executar_trabalho,))
Monitore futuros e feche o executor durante shutdown. Consulte concurrent.futures no Python.
Shutdown
Mantenha uma flag ou evento de parada. Pare de criar recorrências, cancele eventos pendentes quando apropriado e aguarde trabalho já enviado.
Um scheduler em memória não garante que tarefas pendentes sobrevivam ao encerramento.
Relógio civil
Para executar “às 09:00 locais”, converta o horário civil em um atraso ou use um sistema que trate timezone, horário de verão e persistência.
Não use time.monotonic() como timestamp de calendário. Ele serve para medir intervalos.
Testes determinísticos
Injete funções de tempo e delay falsas.
class Relogio:
def __init__(self):
self.agora = 0
def time(self):
return self.agora
def sleep(self, atraso):
self.agora += atraso
relogio = Relogio()
teste = sched.scheduler(relogio.time, relogio.sleep)
Assim, testes não dependem de sleeps reais.
Persistência
Se eventos precisam sobreviver a restart, salve a intenção em banco ou fila durável e reconstrua o scheduler na inicialização.
Não serialize callbacks arbitrários. Persista tipo de tarefa e parâmetros validados.
Segurança
Não permita que usuários agendem callbacks Python arbitrários. Mapeie comandos aprovados para funções conhecidas.
Aplique limites de quantidade, frequência, prioridade e tamanho dos argumentos.
Observabilidade
Registre ID lógico, instante planejado, início real, atraso, duração, resultado e tentativa.
Métricas de fila, atraso e falha mostram quando callbacks estão bloqueando o scheduler.
Quando usar outra ferramenta
Use cron ou Task Scheduler para processos independentes, asyncio para aplicações assíncronas, filas distribuídas para trabalho durável e bibliotecas de scheduling para calendários complexos.
sched é ideal para uma fila temporal local e pequena.
Erros comuns
Os erros mais frequentes são misturar relógios, executar callbacks longos, reagendar com drift, presumir persistência, ignorar exceções, esquecer cancelamento, usar prioridade como preempção e depender de sleeps reais nos testes.
Conclusão
sched oferece scheduling local baseado em uma fila de eventos. Use relógio monotônico para intervalos, eventos absolutos para cadência fixa, prioridades para desempate e callbacks curtos.
Adicione cancelamento, observabilidade e política de falhas conforme o caso. Consulte a documentação oficial de sched e concurrent.futures no Python.
