FastAPI에서 APScheduler로 주기 작업 구현하기

7월 08, 2026

백그라운드에서 일정 간격으로 도는 작업(주기 수집, 정리 배치 등)이 필요할 때, 별도의 크론 서버 없이 FastAPI 앱 안에서 처리하는 방법을 정리합니다. 실제 운영 중인 유튜브 쇼츠 자동화 서버에서 트렌드 수집을 주기적으로 돌리는 코드를 기준으로 설명합니다.

이 글의 가정: FastAPI 앱이 이미 동작하고 있고, Python 3.10 이상 환경을 가정합니다. 패키지는 pip install apscheduler로 설치합니다.

APScheduler란

APScheduler(Advanced Python Scheduler, 파이썬에서 예약 작업을 관리하는 라이브러리)는 "언제 무엇을 실행할지"를 코드로 등록해두면 그 시점에 함수를 대신 호출해 줍니다. FastAPI가 비동기(async) 기반이므로, 같은 이벤트 루프에서 도는 AsyncIOScheduler를 사용합니다.

앱 생명주기에 스케줄러 붙이기

스케줄러는 앱이 켜질 때 시작하고, 꺼질 때 정리해야 합니다. FastAPI의 startup / shutdown 이벤트에 연결합니다.

from apscheduler.schedulers.asyncio import AsyncIOScheduler

scheduler = AsyncIOScheduler(timezone="Asia/Seoul")

@app.on_event("startup")
async def startup():
    scheduler.start()

@app.on_event("shutdown")
async def shutdown():
    scheduler.shutdown()

timezone을 명시하면 서버 로캘과 무관하게 의도한 시간대로 동작합니다.

interval 작업 등록하기

가장 흔한 형태는 "일정 간격마다 반복(interval)"입니다. 수집 함수를 일정 주기로 등록합니다.

@app.on_event("startup")
async def startup():
    scheduler.add_job(run_collection, "interval", minutes=30, id="collect")
    scheduler.start()

여기서는 약 30분 간격을 사용합니다. 이 값은 운영상 무난한 주기로 쓰는 예시이며, 적정 간격은 데이터가 얼마나 자주 바뀌는지와 외부 API 호출 한도에 맞춰 정하는 것이 일반적인 권장 방식입니다. id를 지정하면 등록된 작업을 식별하고 관리하기 쉽습니다.

부팅 직후 한 번 실행하기

interval 작업은 다음 주기가 되어야 처음 실행됩니다. 앱을 켜자마자 한 번 돌리고 싶다면 asyncio.create_task로 즉시 실행을 함께 걸어 둡니다.

import asyncio

@app.on_event("startup")
async def startup():
    scheduler.add_job(run_collection, "interval", minutes=30, id="collect")
    scheduler.start()
    asyncio.create_task(run_collection())  # 부팅 직후 1회

이렇게 하면 "지금 한 번 + 이후 주기적으로"가 자연스럽게 구성됩니다. 같은 방식으로 서로 다른 id를 가진 여러 작업을 하나의 스케줄러에 등록할 수 있습니다.

주의할 점

여러 프로세스로 띄우면 중복 실행됩니다. 워커를 여러 개(예: Gunicorn 다중 워커)로 실행하면 각 프로세스가 같은 작업을 각자 돌립니다. 단일 프로세스로 운영하거나, 다중 인스턴스라면 실행 주체를 하나로 제한하는 장치가 별도로 필요합니다.
작업이 겹치지 않게 하려면 한 번의 실행이 다음 주기보다 오래 걸리지 않는지 확인하는 것이 안전합니다.
@app.on_event는 오래된 방식이며, 최신 FastAPI에서는 lifespan 방식도 제공합니다. 여기서는 실제 코드 기준으로 on_event를 사용했습니다.

마무리

정리하면, AsyncIOScheduler를 앱 생명주기에 연결하고 interval 작업을 등록하는 것만으로 별도 크론 없이 주기 작업을 돌릴 수 있습니다. 이 스케줄러가 호출하는 수집 함수 내부가 외부 API를 어떻게 비동기로 부르는지는 httpx AsyncClient로 외부 API 비동기 호출하기에서 이어집니다. FastAPI와 데이터베이스 연결의 기본 골격은 FastAPI와 PostgreSQL 연동하기를, 수집 작업이 다루는 API 한도 관리는 YouTube Data API 쿼터 초과를 피하는 방법을 참고하세요.

이 글이 도움이 됐다면 구독과 공유 부탁드립니다.

이 블로그 검색

AI 자동화 연구소