Como Agendar Tarefas em Python Sem Celery e Alternativas
Todo desenvolvedor backend provavelmente já enfrentou isso. Você está construindo um serviço web com FastAPI ou Flask, o projeto ainda é pequeno, rodando em um único servidor. Então o product manager pergunta: "Ei, vamos enviar um resumo para os clientes todas as manhãs às 9h, e resetar carrinhos expirados a cada hora?"
O primeiro instinto é adicionar Celery com Celery Beat e Redis. Mas subir um broker, configurar workers e monitorar daemons só para duas tarefas em background parece um exagero. A segunda opção é o cron do sistema, mas isso espalha a lógica entre o código da aplicação e a configuração do servidor. Você quer o scheduler embutido no seu código, capaz de lidar com asyncio, e escalando entre múltiplos nodes quando necessário.
É aqui que a biblioteca APScheduler (Advanced Python Scheduler) se mostra útil.
O que a biblioteca pode fazer
O APScheduler existe há um tempo, e o projeto está atualmente em transição para a versão 4.0. A quarta branch ainda está em status pré-release, mas restructurou significativamente a arquitetura da biblioteca.
Antes era principalmente um scheduler local para um único processo, mas agora a ferramenta evoluiu para um sistema completo de filas e agendamento distribuído. O caso de uso mais simples ainda roda em apenas três linhas de código.
O conceito central: você declara funções Python regulares ou coroutines e anexa um trigger com as condições de execução necessárias.
from apscheduler.schedulers.asyncio import AsyncIOScheduler
scheduler = AsyncIOScheduler()
async def send_digest():
print("Отправляем утренний дайджест...")
# Запуск каждый будний день в 9 утра
scheduler.add_job(send_digest, 'cron', day_of_week='mon-fri', hour=9, minute=0)
scheduler.start()
A biblioteca gerencia o rastreamento de tempo, calcula offsets e invoca funções no momento certo.
Opções de agendamento
Quatro tipos de trigger nativos estão disponíveis:
- Trigger cron. Sintaxe estilo Linux familiar. Você pode definir dias da semana, meses, horas e minutos específicos de forma flexível.
- Intervalos. Rode a cada N segundos, minutos ou horas. Útil para polling regular de APIs externas.
- Trigger de calendário. Necessário quando o intervalo depende do tamanho de um mês ou ano. Por exemplo, rode uma tarefa estritamente no primeiro dia de cada mês ao meio-dia.
- Execução única. Dispara exatamente uma vez em uma data e hora futura especificada.
Se as condições nativas não forem suficientes, triggers podem ser combinados usando regras compostas ou você pode escrever sua própria classe com lógica customizada.
Proteção contra falhas comuns
Na prática, tarefas em background frequentemente encontram sobrecargas e atrasos. O APScheduler inclui vários mecanismos úteis que são frequentemente negligenciados em soluções customizadas construídas sobre while True e sleep.
Primeiro, limitação de execuções concorrentes. Se você configurou uma exportação de relatório pesado a cada cinco minutos, mas a execução anterior travou por sete minutos, o scheduler não vai iniciar uma segunda instância em paralelo e não vai sobrecarregar o banco de dados.
Segundo, o parâmetro jitter. Ele adiciona um delay aleatório ao horário de início da tarefa. Imagine que você tem cem workers, e todos precisam atualizar o cache exatamente às 00:00. Sem jitter, o banco de dados receberá um pico de carga instantâneo. Com um offset aleatório de alguns segundos, a carga é distribuída uniformemente.
Terceiro, tratamento de grace time para tarefas atrasadas. Se o servidor congelou sob carga ou estava reiniciando, o scheduler verifica quão atrasada a tarefa está. Se o delay está dentro do limite aceitável, a tarefa será executada; se não, será ignorada sem acúmulo na fila.
Armazenamento e modo distribuído
Para scripts simples, as tarefas podem ser mantidas em memória. Mas se o serviço reiniciar, o schedule será resetado. Para evitar isso, a biblioteca suporta armazenamento persistente:
- PostgreSQL
- MySQL
- SQLite
- MongoDB
Na versão quatro, o scheduler aprendeu a trabalhar em um cluster distribuído. Múltiplas instâncias da aplicação se conectam a um banco de dados compartilhado e broker de eventos (Redis, PostgreSQL LISTEN/NOTIFY e MQTT são suportados).
Isso permite escalabilidade horizontal: um node cai, os outros pegam as tarefas da fila.
# Пример концепции работы с постоянным хранилищем
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
jobstores = {
'default': SQLAlchemyJobStore(url='postgresql+asyncpg://user:pass@localhost/mydb')
}
scheduler = AsyncIOScheduler(jobstores=jobstores)
Código síncrono e assíncrono
A biblioteca se encaixa bem na stack Python moderna. Diferentes implementações de scheduler estão disponíveis:
AsyncIOSchedulere integração com Trio para backends assíncronos modernos como FastAPI, Litestar ou Aiohttp.BackgroundSchedulereBlockingSchedulerpara scripts síncronos, Django e Flask.
Você não precisa fazer workarounds síncronos para funções async com asyncio.run() — o scheduler nativamente aguarda coroutines dentro do event loop principal.
Onde isso é útil
Normalmente recorro ao APScheduler em três situações típicas:
- Microserviços pequenos e bots. Quando fazer deploy de Celery ou RQ seria excessivo, e tarefas periódicas são necessárias diretamente dentro do processo.
- Ações adiadas do usuário. Por exemplo, envie um email pedindo para avaliar um pedido exatamente 24 horas após a entrega.
- Limpeza de dados temporários e sincronização periódica de dados de referência de sistemas externos.
Vale a pena usar em um projeto
Se você está escrevendo em Python e precisa de execução de tarefas baseada em tempo previsível, o APScheduler é uma das opções mais maduras.
A única nuance no momento: o período de transição entre as versões 3.x e 4.0. A terceira branch foi testada em produção por anos, é maximamente estável, mas tem limitações em operação distribuída. A versão 4.0 traz arquitetura moderna e escalabilidade, mas o autor honestamente alerta sobre possíveis mudanças na API antes do release final.
Para produção atual, é mais seguro ficar com a branch estável 3.x, e a versão 4.0 vale a pena testar em projetos pessoais ou novos microserviços enquanto acompanha as mudanças que estão por vir.
Projetos relacionados