4단계 — APScheduler 로 정기 작업

"매일 오전 9시 통계 리포트", "5분마다 외부 API 동기화" — 이런 정기 작업 이 백엔드에 항상 필요해요.

APScheduler 설치 + 첫 작업

uv add apscheduler

# schedulers/daily_report.py
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger

def daily_report():
    print(f"[{datetime.now()}] 일일 리포트 생성")
    # … 실제 로직

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

main.py 에서 시작/종료

from contextlib import asynccontextmanager
from schedulers.daily_report import scheduler

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

app = FastAPI(lifespan=lifespan)

lifespan 은 FastAPI 의 시작/종료 훅. 서버가 뜰 때 스케줄러도 같이 떠요.

트리거 4종

트리거	용도	예시
CronTrigger	시간표 기반	매일 09:00 KST
IntervalTrigger	일정 간격	5분마다
DateTrigger	한 번만	2026-12-31 23:59
CombiningTrigger	여러 조건	평일 오전만

from apscheduler.triggers.interval import IntervalTrigger

scheduler.add_job(sync_external_api, IntervalTrigger(minutes=5))

멱등성이 핵심

스케줄러가 두 번 실행 되어도 안전해야 해요. 같은 데이터를 다시 INSERT 하면 에러 나거나 중복.

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 가 멱등의 핵심 도구.

직접 해 보기

매 분 실행되는 작업을 등록해 콘솔에 시각을 출력해 보세요. IntervalTrigger(minutes=1) + print(datetime.now()). 1분 기다린 후 출력이 보이면 성공.

더 깊이

• APScheduler 노트

다음 단계

5단계에서는 외부 API 를 예의 바르게 호출하는 크롤러 윤리를 배워요.