APScheduler + KST 스케줄

정기 작업 예약의 표준 Python 라이브러리. cron 보다 코드 안에서 관리하기 편함.

1. 설치

uv add apscheduler

2. AsyncIOScheduler 기본

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

KST = pytz.timezone("Asia/Seoul")

scheduler = AsyncIOScheduler(timezone=KST)

@scheduler.scheduled_job(CronTrigger(hour=3, minute=0), id="daily-nps-crawl")
async def daily_crawl():
    await crawl_nps()

scheduler.start()

서버 시간대 무관하게 한국 시간으로 실행.

3. 트리거 종류

# 매일 03:00 KST
CronTrigger(hour=3, minute=0)

# 매시간
IntervalTrigger(hours=1)

# 특정 시각 1회
DateTrigger(run_date=datetime(2026, 5, 10, 9, 0))

# 매주 월 · 수 · 금
CronTrigger(day_of_week="mon,wed,fri", hour=3)

4. 멱등 옵션 — 중복 실행 방어

JOB_DEFAULTS = {
    "max_instances": 1,           # 같은 잡 동시 실행 1 개
    "coalesce": True,              # 놓친 실행 병합
    "misfire_grace_time": 300,     # 5분 이내 지연은 실행
    "replace_existing": True,      # 재등록 시 덮어쓰기
}

scheduler = AsyncIOScheduler(timezone=KST, job_defaults=JOB_DEFAULTS)

이 4 가지 조합이 멱등성 SSOT.

• max_instances=1 — 긴 작업이 다음 트리거와 겹쳐도 한 번만
• coalesce=True — 서버 다운 복구 후 밀린 N 번을 1 번으로
• misfire_grace_time — 짧은 지연은 허용
• replace_existing=True — 앱 재시작 시 잡 중복 방지

5. FastAPI lifespan 에 등록

from contextlib import asynccontextmanager
from fastapi import FastAPI

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

app = FastAPI(lifespan=lifespan)

앱 시작 시 자동 시작 · 종료 시 진행 중 잡 완료 기다림.

6. 수동 트리거 (관리 UI)

@app.post("/admin/jobs/{job_id}/run")
async def trigger_job(job_id: str):
    job = scheduler.get_job(job_id)
    if not job:
        raise HTTPException(404)
    job.modify(next_run_time=datetime.now(KST))
    return {"ok": True}

운영자가 UI 에서 "지금 실행" 버튼. 다음 cron 기다리지 않고 즉시.

7. 중복 실행 방지 — 분산 락

APScheduler 자체는 한 프로세스 가정. 여러 인스턴스 배포 시 같은 잡 중복.

async def crawl_with_lock():
    async with redis_lock("lock:daily-nps-crawl", ttl=3600):
        await crawl_nps()

Redis SET NX EX 로 글로벌 락. 먼저 획득한 인스턴스만 실행.

8. jobstore — 상태 영속화

기본은 메모리. 앱 재시작 시 다음 트리거 재계산.

from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore

scheduler = AsyncIOScheduler(
    jobstores={"default": SQLAlchemyJobStore(url="postgresql://...")},
    timezone=KST,
)

DB 에 잡 정의 저장. 단점: 복잡 · 재시작 드물면 과잉.

대부분 프로젝트는 메모리 jobstore + 코드로 등록 (decorator) 이 간단.

9. 실행 결과 기록

@scheduler.scheduled_job(CronTrigger(hour=3), id="nps")
async def nps_job():
    start = time.time()
    try:
        rows = await crawl_nps()
        await db.execute(
            "INSERT INTO crawl_runs (source, status, rows, duration_ms) VALUES ($1, $2, $3, $4)",
            "nps", "ok", rows, int((time.time() - start) * 1000)
        )
    except Exception as e:
        await db.execute(
            "INSERT INTO crawl_runs (source, status, error) VALUES ($1, $2, $3)",
            "nps", "fail", str(e)
        )
        raise

실행 이력 테이블이 대시보드 · 알림의 기반.

10. 자주 걸리는 자리

• timezone 누락 — UTC 에서 실행 · 의도와 다른 시각
• misfire_grace_time 기본 너무 짧음 — 1초 지연도 skip
• 여러 인스턴스 동시 실행 — 분산 락 없이는 중복
• async 잡을 동기 scheduler 에 — 차단

하고픈 말

APScheduler + KST + 멱등 4 옵션이 Python 백엔드 스케줄링의 기본 세트. cron 파일 · systemd timer 보다 코드 관리 · 테스트 용이.

Next

• 05-incremental-dedup