4단계
4단계 — APScheduler 로 정기 작업
30 분
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분 기다린 후 출력이 보이면 성공.
더 깊이
다음 단계
5단계에서는 외부 API 를 예의 바르게 호출하는 크롤러 윤리를 배워요.