---
title: "APScheduler em Python: agende tarefas sem Airflow"
url: "https://python.dev.br/blog/apscheduler-agendar-tarefas-python/"
markdown_url: "https://python.dev.br/blog/apscheduler-agendar-tarefas-python.MD"
description: "Aprenda APScheduler em Python para agendar jobs em processo com cron, intervalo e data, persistir estado em SQLAlchemy e escolher entre schedule, Celery e Airflow."
date: "2026-09-17"
author: "Equipe Python Dev BR"
---

# APScheduler em Python: agende tarefas sem Airflow

Aprenda APScheduler em Python para agendar jobs em processo com cron, intervalo e data, persistir estado em SQLAlchemy e escolher entre schedule, Celery e 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`](/blog/automatizacao-com-python/) só para scripts curtos; e escolha [Celery](/blog/fastapi-background-tasks-celery-redis-2026/) ou [Airflow](/blog/airflow-python-orquestracao-pipelines/) 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](/blog/virtual-environments-python/) ou [`uv`](/blog/uv-gerenciador-pacotes-python/)):

```bash
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:

```bash
pip install "apscheduler[sqlalchemy]" sqlalchemy
```

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

## Exemplo mínimo: job a cada minuto

```python
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

| Peça | Papel | Exemplos |
| --- | --- | --- |
| **Trigger** | decide *quando* disparar | `date`, `interval`, `cron` |
| **Job** | a função/callable a executar | `verificar_fila_pix` |
| **Executor** | *como* rodar (thread, processo, asyncio) | `ThreadPoolExecutor`, `ProcessPoolExecutor` |
| **Job store** | *onde* guardar o agendamento | memória, SQLAlchemy, MongoDB |
| **Scheduler** | orquestra tudo e controla start/shutdown | `BlockingScheduler`, `BackgroundScheduler`, `AsyncIOScheduler` |

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`)

```python
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`)

```python
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

```python
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:

```python
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](/blog/python-dotenv-env-vars-config/) e do módulo [`secrets`](/blog/python-secrets-tokens-senhas-seguros/) valem aqui.

## Integração com FastAPI (lifespan)

```python
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](/blog/alerta-vagas-python-telegram/): em vez de depender só do cron do GitHub Actions, o worker local (ou um container sempre ligado) agenda a coleta.

```python
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

| Critério | `schedule` | **APScheduler** | Celery | Airflow |
| --- | --- | --- | --- | --- |
| Instalação / ops | mínima | baixa | média (broker) | alta (scheduler, webserver, DB) |
| Cron real | limitado | sim | via Celery Beat | sim (DAGs) |
| Persistência de agenda | não | store opcional | broker + result backend | metadados no DB |
| Distribuição entre máquinas | não | limitada | forte | forte |
| Melhor para | script único | app/worker leve | filas e retries | pipelines de dados |
| Exemplo no site | [automação](/blog/automatizacao-com-python/) | este guia | [FastAPI + Celery](/blog/fastapi-background-tasks-celery-redis-2026/) | [Airflow](/blog/airflow-python-orquestracao-pipelines/) |

**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”:

```python
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](/blog/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.