Nem toda tarefa agendada precisa de um serviço externo, uma fila distribuída ou um cron do sistema. Para scripts, simuladores, testes e processos locais que precisam executar funções em horários definidos, o módulo sched no Python oferece um agendador pequeno e previsível dentro da biblioteca padrão. Ele organiza eventos por instante e prioridade, espera pelo momento adequado e chama funções com argumentos definidos pelo programa.
Neste guia, você aprenderá a criar um scheduler, usar enter() e enterabs(), cancelar eventos, executar em modo bloqueante ou não bloqueante e implementar repetições seguras. O conteúdo complementa nossos artigos sobre atexit no Python, tracebacks, contextvars, collections e diagnóstico de scripts lentos.
O que o módulo sched resolve
Um agendador recebe eventos e decide quando cada ação deve ser executada. Em sched, tudo acontece dentro do processo atual. Não há persistência automática, banco de dados, workers remotos ou recuperação após reinício.
Isso torna o módulo apropriado para:
- executar callbacks depois de um atraso;
- simular eventos em testes;
- programar tarefas curtas dentro de uma ferramenta local;
- organizar etapas temporizadas de uma demonstração;
- criar um pequeno loop periódico sem dependências.
Para tarefas de produção que precisam sobreviver a crashes, distribuir carga ou executar em várias máquinas, use uma solução persistente e observável.
Criar um scheduler
A classe principal é sched.scheduler.
import sched
agendador = sched.scheduler()Nas versões atuais, os argumentos padrão são time.monotonic para medir o tempo e time.sleep para esperar. Isso evita que ajustes no relógio civil atrasem ou antecipem eventos relativos.
A documentação oficial de sched informa que a classe pode ser usada com segurança em ambientes multithread desde Python 3.3, embora a lógica da aplicação ainda precise proteger seus próprios dados compartilhados.
Agendar depois de um atraso
enter() recebe um atraso em unidades da função de tempo, uma prioridade, uma ação, uma tupla de argumentos e, opcionalmente, argumentos nomeados.
import sched
agendador = sched.scheduler()
def avisar(mensagem):
print(mensagem)
agendador.enter(
delay=2,
priority=1,
action=avisar,
argument=("Dois segundos se passaram",),
)
agendador.run()Com as funções padrão, o atraso é expresso em segundos. O método devolve um objeto de evento que pode ser usado posteriormente para cancelamento.
Argumentos posicionais e nomeados
O parâmetro argument deve ser uma sequência, normalmente uma tupla. kwargs recebe um dicionário.
def registrar(nome, status="ok"):
print(nome, status)
agendador.enter(
1,
1,
registrar,
argument=("importação",),
kwargs={"status": "concluída"},
)Evite usar objetos mutáveis que serão modificados antes da execução sem uma intenção clara. O callback receberá as referências armazenadas no evento.
Agendar em um instante absoluto
enterabs() agenda com base em um valor absoluto produzido pela função de tempo configurada.
import time
agendador = sched.scheduler(
timefunc=time.monotonic,
delayfunc=time.sleep,
)
instante = time.monotonic() + 5
agendador.enterabs(
instante,
1,
avisar,
argument=("Executado no prazo monotônico",),
)Não misture valores de time.time() com um scheduler baseado em time.monotonic(). Os dois relógios possuem origens e finalidades diferentes.
Relógio monotônico versus relógio civil
time.monotonic() só avança e não é afetado por mudanças manuais, sincronização NTP ou horário de verão. É a melhor opção para atrasos e intervalos.
time.time() representa uma escala próxima do horário Unix e pode ser necessário quando o requisito é “às 15:00 de uma data”. Nesse caso, converta a data cuidadosamente e esteja preparado para alterações do relógio.
A documentação oficial de time explica as diferenças entre os relógios. Para calendários com fusos, considere datetime e zoneinfo antes de chegar ao scheduler.
Prioridade dos eventos
Quando dois eventos possuem o mesmo horário, a prioridade define a ordem. Números menores são executados primeiro.
agendador.enter(1, 20, avisar, argument=("prioridade 20",))
agendador.enter(1, 5, avisar, argument=("prioridade 5",))
agendador.enter(1, 10, avisar, argument=("prioridade 10",))A prioridade não interrompe uma ação que já começou. Ela apenas organiza eventos que estão aguardando.
Executar a fila
run() processa eventos na ordem adequada.
agendador.run(blocking=True)No modo bloqueante, o método espera até que todos os eventos existentes sejam executados. A função de atraso é chamada com o tempo restante antes de cada evento.
Eventos atrasados não são descartados
Se uma ação demora mais que o intervalo até o próximo evento, o scheduler fica atrasado. Ele executa os eventos vencidos assim que possível, mantendo a ordem, mas não descarta nenhum deles.
import time
def tarefa_lenta():
time.sleep(3)
print("tarefa lenta terminou")
agendador.enter(0, 1, tarefa_lenta)
agendador.enter(1, 1, avisar, argument=("evento atrasado",))
agendador.run()Se o atraso acumulado é inaceitável, mova tarefas longas para workers controlados ou faça o callback apenas encaminhar trabalho.
Modo não bloqueante
Com blocking=False, run() executa os eventos já vencidos e retorna o prazo do próximo evento ou None quando a fila está vazia.
proximo = agendador.run(blocking=False)
if proximo is None:
print("Fila vazia")
else:
print("Próximo prazo:", proximo)Esse modo permite integrar sched a outro loop. O valor retornado deve ser interpretado conforme a versão e a função de tempo; consulte a documentação da versão usada e calcule o atraso de forma consistente.
Integrar a um loop próprio
import time
while not agendador.empty():
prazo = agendador.run(blocking=False)
if prazo is None:
break
agora = time.monotonic()
atraso = max(0, prazo - agora)
processar_interface_por_ate(min(atraso, 0.1))Um loop gráfico ou assíncrono provavelmente possui seu próprio mecanismo de timers. Nesses casos, adapte o prazo em vez de chamar sleep() na thread da interface.
Consultar a fila
A propriedade queue devolve os eventos pendentes em ordem de execução.
for evento in agendador.queue:
print(
evento.time,
evento.priority,
evento.action,
evento.argument,
evento.kwargs,
)A lista devolvida representa um snapshot ordenado. Alterar essa lista não muda a fila interna.
Verificar se a fila está vazia
empty() indica se existem eventos pendentes.
if agendador.empty():
print("Nada agendado")Em ambiente multithread, o estado pode mudar logo após a consulta. Não use empty() como garantia duradoura sem coordenar as threads.
Cancelar um evento
Guarde o objeto retornado por enter() ou enterabs() e passe-o a cancel().
evento = agendador.enter(
30,
1,
avisar,
argument=("não será executado",),
)
agendador.cancel(evento)Se o evento não está mais na fila, cancel() lança ValueError.
Cancelamento concorrente
try:
agendador.cancel(evento)
except ValueError:
print("O evento já executou ou foi cancelado")Em outra thread, o evento pode começar entre a decisão de cancelar e a chamada. O callback deve ser capaz de verificar um estado de cancelamento próprio quando a operação exige garantia adicional.
Repetir uma tarefa
Sched não possui um tipo especial de tarefa recorrente. O callback agenda a próxima ocorrência.
def executar_periodicamente(intervalo):
print("executando")
agendador.enter(
intervalo,
1,
executar_periodicamente,
argument=(intervalo,),
)
agendador.enter(0, 1, executar_periodicamente, argument=(5,))Esse modelo mede o intervalo a partir do momento do reagendamento. Se a função leva dois segundos, o início seguinte pode ocorrer sete segundos depois do anterior.
Evitar drift em repetições
Para manter uma grade absoluta, calcule o próximo instante a partir do prazo anterior.
def periodica(proximo, intervalo):
print("execução periódica")
seguinte = proximo + intervalo
agendador.enterabs(
seguinte,
1,
periodica,
argument=(seguinte, intervalo),
)
inicio = time.monotonic() + 1
agendador.enterabs(inicio, 1, periodica, argument=(inicio, 5))Defina o que fazer quando várias ocorrências ficaram para trás: executar todas, pular algumas ou recalcular a próxima futura.
Exceções em callbacks
Se uma ação gera uma exceção, run() propaga o erro. O estado interno permanece consistente e o evento que falhou não é executado novamente automaticamente.
def protegido():
try:
operacao()
except Exception:
logger.exception("Evento agendado falhou")
agendador.enter(1, 1, protegido)Escolha entre propagar e capturar conforme a responsabilidade da aplicação. Não esconda falhas críticas sem observabilidade.
Uso com threads
A fila do scheduler é thread-safe, então uma thread pode adicionar eventos enquanto outra executa run(). Seus callbacks e objetos compartilhados não recebem proteção automática.
from threading import Lock
lock = Lock()
estado = {}
def atualizar(chave, valor):
with lock:
estado[chave] = valorTambém considere que um run(blocking=True) já adormecido pode não acordar imediatamente quando outra thread insere um evento anterior, dependendo do padrão de integração. Loops responsivos devem usar verificações curtas ou um mecanismo de sinalização próprio.
Funções de tempo personalizadas
O construtor aceita funções diferentes, o que é excelente para testes.
class RelogioFalso:
def __init__(self):
self.agora = 0.0
def tempo(self):
return self.agora
def esperar(self, segundos):
self.agora += segundos
relogio = RelogioFalso()
agendador = sched.scheduler(relogio.tempo, relogio.esperar)Assim, testes avançam virtualmente sem aguardar segundos reais.
delayfunc recebe zero
Depois de cada evento, o scheduler chama a função de atraso com zero para permitir que outras threads executem.
Uma delayfunc personalizada deve aceitar zero e não tratar esse valor como erro.
Testes determinísticos
eventos = []
agendador.enter(5, 1, eventos.append, argument=("A",))
agendador.enter(2, 1, eventos.append, argument=("B",))
agendador.run()
assert eventos == ["B", "A"]
assert relogio.agora == 5Teste mesma hora com prioridades diferentes, cancelamento, exceção e eventos adicionados durante a execução.
Persistência e reinício
A fila vive apenas na memória. Ao encerrar o processo, todos os eventos pendentes desaparecem.
Para tarefas que devem sobreviver a reinícios, armazene-as em banco ou fila e recrie o scheduler na inicialização. Evite serializar callables arbitrários com pickle.
Várias máquinas
Sched não coordena locks distribuídos nem impede execução duplicada em duas instâncias. Cada processo possui sua própria fila.
Use um scheduler distribuído, banco com leasing ou fila de tarefas quando apenas uma instância deve executar o trabalho.
Observabilidade
Registre ao menos:
- identificador lógico do evento;
- horário planejado e real;
- atraso acumulado;
- duração;
- status e exceção;
- quantidade de eventos pendentes.
Não registre argumentos que contenham tokens, senhas ou dados pessoais.
Encerramento controlado
Use uma flag para impedir novos reagendamentos, cancele eventos pendentes quando apropriado e finalize callbacks em execução.
encerrando = False
def recorrente(intervalo):
executar()
if not encerrando:
agendador.enter(intervalo, 1, recorrente, argument=(intervalo,))Combine o protocolo com sinais e ciclo de vida da aplicação. atexit pode registrar uma última métrica, mas não garante execução em crash.
Erros frequentes
- Misturar
time.time()etime.monotonic(). - Executar callbacks longos na thread do scheduler.
- Presumir que eventos atrasados serão descartados.
- Cancelar sem tratar
ValueError. - Criar repetição acumulando drift sem intenção.
- Usar a fila em memória para tarefas críticas.
- Compartilhar estado sem locks.
- Tratar sched como scheduler distribuído.
Boas práticas
- Use relógio monotônico para intervalos.
- Mantenha callbacks curtos e observáveis.
- Guarde os objetos de evento para cancelamento.
- Defina política para atrasos e repetições.
- Use relógio falso em testes.
- Proteja dados compartilhados.
- Persista tarefas que devem sobreviver.
- Escolha outra solução para múltiplas máquinas.
Conclusão
O módulo sched no Python oferece uma fila temporal simples para executar funções por atraso ou instante absoluto. Ele ordena eventos por horário e prioridade, suporta cancelamento, permite inspeção da fila e funciona com relógios personalizados.
Seu valor está na simplicidade local. Quando callbacks são curtos, o relógio é escolhido corretamente e a aplicação define políticas de atraso, repetição e encerramento, sched resolve timers internos sem dependências. Persistência, distribuição e recuperação, porém, exigem uma infraestrutura diferente.
