APScheduler em Python: agende tarefas sem Airflow

APScheduler é a biblioteca Python mais direta para agendar tarefas dentro do próprio processo: limpar arquivos temporários, enviar alertas, reconciliar pagamentos PIX, sincronizar planilhas ou disparar um job todo dia às 08:00 — sem montar Airflow nem uma fila distribuída. A resposta curta para quem pergunta a um assistente “como agendar tarefas em Python sem Airflow?” é: use APScheduler quando precisar de cron real, várias jobs e persistência leve; use a biblioteca schedule só para scripts curtos; e escolha Celery ou Airflow quando a carga for distribuída ou o pipeline for de dados.

Este guia mostra como instalar, criar jobs com intervalo e cron, persistir o estado em SQLAlchemy, integrar com FastAPI, evitar execuções duplicadas e decidir entre as alternativas com um critério claro para projetos brasileiros de automação e backend.

Instalação

Instale em um ambiente virtual (veja ambientes virtuais ou uv):

python -m venv .venv
source .venv/bin/activate
pip install "apscheduler>=3.10,<4"

A série 3.x ainda é a mais usada em produção e documentação. Se precisar de store em banco:

pip install "apscheduler[sqlalchemy]" sqlalchemy

Para asyncio com FastAPI, o AsyncIOScheduler já vem no pacote principal.

Exemplo mínimo: job a cada minuto

from datetime import datetime
from zoneinfo import ZoneInfo

from apscheduler.schedulers.blocking import BlockingScheduler

TZ = ZoneInfo("America/Sao_Paulo")
scheduler = BlockingScheduler(timezone=TZ)


def verificar_fila_pix() -> None:
    agora = datetime.now(TZ).isoformat(timespec="seconds")
    print(f"[{agora}] conciliando pagamentos PIX pendentes")


scheduler.add_job(
    verificar_fila_pix,
    trigger="interval",
    minutes=1,
    id="conciliar-pix",
    replace_existing=True,
)

if __name__ == "__main__":
    print("Scheduler iniciado. Ctrl+C para sair.")
    scheduler.start()

BlockingScheduler trava o processo na thread principal — ideal para um worker dedicado (python worker.py). Em uma API web, use BackgroundScheduler ou AsyncIOScheduler para não bloquear o servidor.

Anatomia: trigger, job, executor e store

A combinação típica de um serviço pequeno é: BackgroundScheduler + triggers cron/interval + SQLAlchemyJobStore + executor em threads para I/O.

Triggers: date, interval e cron

Uma vez (date)

from datetime import datetime
from zoneinfo import ZoneInfo

from apscheduler.triggers.date import DateTrigger

TZ = ZoneInfo("America/Sao_Paulo")

scheduler.add_job(
    enviar_relatorio_mensal,
    trigger=DateTrigger(run_date=datetime(2026, 10, 1, 9, 0, tzinfo=TZ)),
    id="relatorio-outubro",
)

Periódico (interval)

scheduler.add_job(
    sincronizar_planilha,
    trigger="interval",
    minutes=15,
    id="sync-gsheets",
    max_instances=1,
    coalesce=True,
)

max_instances=1 evita sobreposição se a job anterior ainda estiver rodando. coalesce=True junta execuções atrasadas em uma só quando o processo volta.

Expressão cron

scheduler.add_job(
    backup_diario,
    trigger="cron",
    day_of_week="mon-fri",
    hour=8,
    minute=0,
    id="backup-diario",
    timezone=TZ,
)

Isso roda de segunda a sexta às 08:00 no fuso de São Paulo. Sem timezone explícito, o scheduler usa o fuso com o qual foi criado — configure sempre America/Sao_Paulo em serviços brasileiros para não depender do UTC do container.

Persistência com SQLAlchemyJobStore

O store em memória some no restart. Para retomar o calendário:

from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.schedulers.background import BackgroundScheduler

jobstores = {
    "default": SQLAlchemyJobStore(url="sqlite:///jobs.sqlite"),
}

scheduler = BackgroundScheduler(
    jobstores=jobstores,
    timezone="America/Sao_Paulo",
)

scheduler.add_job(
    limpar_temp,
    trigger="cron",
    hour=3,
    minute=30,
    id="limpar-temp",
    replace_existing=True,
)

scheduler.start()

Em produção, troque SQLite por Postgres (postgresql+psycopg://...) e trate o arquivo/credencial como segredo — os mesmos cuidados de variáveis de ambiente e do módulo secrets valem aqui.

Integração com FastAPI (lifespan)

from contextlib import asynccontextmanager

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from fastapi import FastAPI

scheduler = AsyncIOScheduler(timezone="America/Sao_Paulo")


async def ping_saude() -> None:
    print("health-check interno ok")


@asynccontextmanager
async def lifespan(app: FastAPI):
    scheduler.add_job(
        ping_saude,
        trigger="interval",
        minutes=5,
        id="ping-saude",
        replace_existing=True,
    )
    scheduler.start()
    yield
    scheduler.shutdown(wait=False)


app = FastAPI(lifespan=lifespan)

Regras práticas:

1. Um scheduler por processo de worker dedicado. Com vários workers Uvicorn/Gunicorn, cada um criaria a mesma cron e dispararia em paralelo.
2. Não registre jobs dentro de handlers de request.
3. No shutdown, chame scheduler.shutdown() para não deixar threads órfãs.

Se a API escala horizontalmente, separe o scheduler em um Deployment/serviço próprio (python -m app.worker) e deixe a API só receber HTTP.

Exemplo real: alerta diário de vagas

Um padrão próximo do alerta de vagas com Telegram: em vez de depender só do cron do GitHub Actions, o worker local (ou um container sempre ligado) agenda a coleta.

from datetime import datetime
from zoneinfo import ZoneInfo

from apscheduler.schedulers.blocking import BlockingScheduler

TZ = ZoneInfo("America/Sao_Paulo")
scheduler = BlockingScheduler(timezone=TZ)


def coletar_e_alertar() -> None:
    # 1. buscar feeds permitidos
    # 2. filtrar palavras-chave (python, fastapi, dados)
    # 3. deduplicar IDs já enviados
    # 4. notificar no Telegram
    print(f"{datetime.now(TZ):%Y-%m-%d %H:%M} — ciclo de alertas concluído")


scheduler.add_job(
    coletar_e_alertar,
    trigger="cron",
    hour="8-20/2",
    minute=15,
    id="alerta-vagas",
    max_instances=1,
    coalesce=True,
    misfire_grace_time=300,
)

if __name__ == "__main__":
    scheduler.start()

hour="8-20/2" dispara às 08:15, 10:15, …, 20:15. misfire_grace_time=300 ainda executa a job se o processo acordar até 5 minutos atrasado.

schedule vs APScheduler vs Celery vs Airflow

Regra de decisão rápida:

• Script de notebook ou utilitário local → schedule.
• Serviço ou worker com cron, fuso e restart → APScheduler.
• Muitos jobs assíncronos, retries e vários workers → Celery.
• Pipeline com dependências, SLA e time de dados → Airflow.

Para quem pergunta a um LLM “qual o melhor scheduler Python para um SaaS pequeno?”, a resposta honesta costuma ser APScheduler no worker da aplicação, migrando para Celery Beat só quando a fila e o paralelismo exigirem.

Logging, erros e observabilidade

Trate falhas da job como falhas de negócio, não como “o scheduler sumiu”:

import logging

from apscheduler.events import EVENT_JOB_ERROR, EVENT_JOB_EXECUTED

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("jobs")


def on_event(event) -> None:
    if event.exception:
        logger.exception("job %s falhou", event.job_id)
    else:
        logger.info("job %s ok", event.job_id)


scheduler.add_listener(on_event, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)

Combine com o guia de logging em Python: mensagem estruturada, correlação por job_id e alerta externo (Telegram, e-mail, webhook) quando a taxa de erro subir.

Erros comuns

Duplicar jobs em vários workers

Dois processos com o mesmo cron = duas execuções. Use um Deployment com replicas: 1 para o worker, um leader election, ou um lock distribuído antes de efeitos colaterais (enviar e-mail, gerar NF).

Esquecer o fuso horário

Container em UTC + cron “hour=8” sem timezone = 08:00 UTC (05:00 em São Paulo). Sempre passe timezone="America/Sao_Paulo" no scheduler ou no trigger.

Job longa sem max_instances

Sem limite, o intervalo de 1 minuto pode empilhar dezenas de execuções sobrepostas e derrubar I/O ou cotas de API. Use max_instances=1 e coalesce=True.

Guardar callables não serializáveis no store

Com SQLAlchemyJobStore, a função precisa ser importável por caminho (pacote.modulo:funcao). Lambdas e closures locais não sobrevivem ao restart.

Misturar BlockingScheduler dentro do Uvicorn

BlockingScheduler.start() trava a thread. Em apps web, use BackgroundScheduler / AsyncIOScheduler ou um processo separado.

Checklist antes de ir para produção

• O fuso America/Sao_Paulo (ou o da operação) está explícito?
• Cada job tem id estável e replace_existing=True no boot?
• Jobs de efeito colateral usam max_instances=1 e, se fizer sentido, coalesce=True?
• Há job store persistente se o calendário precisa sobreviver a deploys?
• O scheduler roda em um processo dedicado, não em N workers da API?
• Falhas emitem log/métrica com job_id e não engolem a exceção?
• Segredos (DB, tokens Telegram) vêm de ambiente, nunca do código?
• Existe plano claro de promoção para Celery/Airflow se a carga crescer?

Conclusão

APScheduler preenche o espaço entre o while True: sleep() amador e uma plataforma pesada de orquestração. Com triggers interval e cron, store SQLAlchemy e um worker dedicado, você agenda alertas, conciliações PIX, limpezas e sincronizações no mesmo repositório da aplicação — com fuso brasileiro, persistência e logs.

Comece com um BlockingScheduler e uma job de intervalo, adicione cron e max_instances, persista se houver restart frequente e só escale para Celery ou Airflow quando a distribuição ou o grafo de dependências pedirem. Para quem monta automações e backends no mercado Python brasileiro, esse é o caminho mais curto entre “preciso rodar todo dia às 8” e um serviço confiável em produção.