Step 4 — APScheduler

"Daily 9am report", "sync external API every 5 minutes" — every backend needs scheduled work.

Install + first job

uv add apscheduler

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger

def daily_report():
    print(f"[{datetime.now()}] generating daily report")

scheduler = AsyncIOScheduler(timezone="Asia/Seoul")
scheduler.add_job(daily_report, CronTrigger(hour=9, minute=0))  # 09:00 KST daily

Wire into FastAPI lifespan

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    scheduler.start()
    yield
    scheduler.shutdown()

app = FastAPI(lifespan=lifespan)

Four trigger types

Idempotency is critical

A scheduled job may run twice (restart, race). Same data twice should not break.

def sync_external_api():
    data = fetch_external()
    with get_conn() as conn, conn.cursor() as cur:
        cur.execute("""
            INSERT INTO sync_logs (key, payload, synced_at)
            VALUES (%s, %s, NOW())
            ON CONFLICT (key) DO UPDATE SET payload = EXCLUDED.payload, synced_at = NOW()
        """, (data["key"], json.dumps(data)))

ON CONFLICT DO UPDATE is the idempotency tool.

Try it

Register a 1-minute job that prints datetime.now(). Wait 60s and confirm.

Next

Step 5 covers crawler ethics for external APIs.