APScheduler agenda funções para execução imediata, futura ou recorrente. Ele funciona bem em aplicações pequenas e serviços dedicados, mas exige decisões explícitas sobre reinícios, concorrência e múltiplas instâncias.

Instalar e escolher o scheduler

Instale a versão adotada pelo projeto e fixe-a no arquivo de dependências:

python -m pip install APScheduler

Os exemplos seguem a API 3.x, comum em muitos projetos. O APScheduler 4 reorganiza conceitos e imports, portanto confira a versão principal instalada. BlockingScheduler ocupa o processo principal e serve para um serviço dedicado. BackgroundScheduler usa uma thread em uma aplicação síncrona de longa duração. Aplicações assíncronas devem usar a integração documentada para sua versão. Inicie o scheduler uma única vez e encerre-o de forma controlada.

Criar um agendamento simples

from datetime import datetime
from zoneinfo import ZoneInfo

from apscheduler.schedulers.blocking import BlockingScheduler


def gerar_relatorio() -> None:
    print("relatório iniciado")


scheduler = BlockingScheduler(timezone=ZoneInfo("America/Sao_Paulo"))
scheduler.add_job(
    gerar_relatorio,
    "cron",
    hour=7,
    minute=30,
    id="relatorio-diario",
    replace_existing=True,
    max_instances=1,
)
scheduler.start()

Use date para uma execução, interval para cadência fixa e cron quando o requisito segue o calendário. Um intervalo de 24 horas mede tempo decorrido; “todos os dias às 7h30” acompanha o calendário local e pode atravessar mudanças de horário oficial.

from datetime import datetime, timedelta

scheduler.add_job(
    enviar_lembrete,
    "date",
    run_date=datetime.now(tz=ZoneInfo("America/Sao_Paulo")) + timedelta(minutes=10),
    args=["fatura-1842"],
)
scheduler.add_job(
    atualizar_painel,
    "interval",
    minutes=15,
    id="atualizar-painel",
    replace_existing=True,
)

Com armazenamento persistente, passe argumentos serializáveis e prefira funções no nível do módulo. Lambdas e funções internas são difíceis de reconstruir após um reinício. Não coloque segredos nos argumentos, pois os valores podem aparecer serializados ou em diagnósticos.

Definir o comportamento em atrasos

O computador pode ficar suspenso, o processo pode parar ou todas as threads podem estar ocupadas. misfire_grace_time informa quanto atraso ainda é aceitável. coalesce=True pode reunir várias ocorrências perdidas em uma execução após a recuperação.

scheduler.add_job(
    importar_cotacoes,
    "cron",
    hour=2,
    id="cotacoes-diarias",
    replace_existing=True,
    max_instances=1,
    misfire_grace_time=900,
    coalesce=True,
)

Consolidar execuções funciona para reconstruir um retrato atual, pois o resultado novo substitui os antigos. Pode ser errado no faturamento, onde cada período precisa ser processado. Modele períodos pendentes como registros duráveis e faça a tarefa reivindicá-los. Um disparo bem-sucedido não comprova que a operação de negócio terminou.

Tornar a tarefa idempotente

Não presuma execução exatamente uma vez. O processo pode concluir uma chamada externa e falhar antes de registrar sucesso. Duas instâncias também podem se sobrepor durante a implantação. Dê uma chave estável a cada operação lógica e faça o destino rejeitar duplicatas ou atualizar o mesmo resultado sem dano.

def fechar_dia_contabil(dia: str) -> None:
    if livro_razao.ja_fechado(dia):
        return

    lancamentos = livro_razao.calcular(dia)
    livro_razao.salvar_fechamento(
        dia, lancamentos, chave_operacao=f"fechamento:{dia}"
    )

Uma restrição única no banco é mais segura que consultar e inserir em duas etapas, pois há uma condição de corrida. Use transações curtas, timeout nas chamadas de rede e um estado final claro. Repetições devem seguir uma política explícita conforme a exceção, nunca um loop infinito escondido.

Persistência e implantação

O armazenamento padrão em memória perde os horários quando o processo termina. Isso é aceitável se o código recria cada agendamento na inicialização com ID estável e replace_existing=True. Um armazenamento persistente ajuda com horários criados dinamicamente, mas traz responsabilidades de conexão, backup, esquema e compatibilidade.

Não suponha que compartilhar um job store coordena schedulers independentes. Persistência não elege um líder. Em servidor web com vários workers, execute um serviço dedicado ou use o agendador da plataforma para enviar trabalho a uma fila. Evite manter as versões antiga e nova ativas juntas além do necessário.

O health check deve diferenciar “processo vivo” de “scheduler acessa o armazenamento e executores avançam”. No encerramento, pare de aceitar trabalho e permita um período compatível com a duração das tarefas.

Observar a execução real

Registre ID, horário previsto, início real, duração, tentativa e resultado. Evite argumentos completos quando contêm dados pessoais. Eventos do APScheduler podem alimentar contadores de sucesso, falha e ocorrências perdidas. Alerte sobre atrasos persistentes e duração anormal, não sobre toda falha transitória.

Em chamadas a outro serviço, use um ID de correlação de ponta a ponta. Mantenha rótulos de métricas limitados: nome estável da tarefa é útil, ID de cliente cria cardinalidade excessiva. Teste exceções, executor atrasado e reinício com ocorrência pendente.

Use o scheduler adequado ao ciclo de vida da aplicação e confira a documentação da versão instalada, pois APIs variam entre gerações principais. Defina o fuso explicitamente e trate horários locais ambíguos durante transições.

Planejar falhas e duplicação

max_instances=1 evita sobreposição no mesmo scheduler, mas não coordena processos independentes. Em servidores web com vários workers, iniciar um scheduler em cada worker pode repetir a tarefa. Prefira um processo dedicado ou um sistema distribuído.

Tarefas devem ser idempotentes, registrar início e resultado, ter timeout e poder ser repetidas conscientemente. Configure tolerância para execuções atrasadas e decida se várias ocorrências perdidas devem ser consolidadas. Para trabalhos distribuídos, compare com Celery e tarefas em background.

APScheduler agenda trabalho, mas não é uma fila distribuída completa. Uma fila costuma ser melhor quando muitas máquinas executam tarefas, há workers especializados, retries duráveis ou fila de falhas, ou o sistema precisa absorver picos fora do processo web. Uma arquitetura comum deixa o scheduler publicar uma mensagem pequena e entrega a operação pesada aos workers.

Para manutenção periódica ou automação interna moderada, APScheduler pode ser mais simples. Documente responsável, fuso, política de atraso, concorrência, chave de idempotência e recuperação. Essas decisões importam mais que a expressão cron.

Checklist antes de publicar

Valide o agendamento em um processo de homologação com o mesmo fuso e configuração de armazenamento. Teste execução normal, falha intencional, tarefa que demora mais que o intervalo e reinício durante o trabalho. Confirme que os logs identificam uma operação lógica sem expor dados e que o operador consegue repeti-la com segurança. Alterar a quantidade de réplicas da aplicação não deve multiplicar execuções.

Defina ainda quem pode modificar horários e como as mudanças são auditadas. Expressões cron fornecidas por usuários precisam de validação e limite razoável de frequência, pois um erro pode gerar carga excessiva. Prefira configuração revisada junto ao código quando os horários forem política operacional.

A documentação oficial do APScheduler, consultada em 22 de julho de 2026, detalha tarefas, schedules, jobs, armazenamento e executores. Fixe a versão usada e valide a API correspondente antes de copiar exemplos entre versões.