APScheduler로 파이썬 고급 스케줄링 하기｜멀티 태스크, DB 트리거까지

기본적인 스케줄링은 schedule 모듈로 충분합니다. 하지만 여러 작업을 동시에 실행하거나, 정밀한 실행 조건(초 단위, 요일 지정, DB 기반 트리거 등)이 필요할 때는? 이럴 땐 APScheduler(Advanced Python Scheduler)를 사용하는 것이 가장 강력한 선택입니다.

이 글에서는 APScheduler의 주요 기능과 실전 예제를 통해, 복잡한 업무 스케줄 관리를 어떻게 파이썬으로 자동화할 수 있는지 알아봅니다.

1. APScheduler란?

APScheduler는 파이썬에서 다양한 유형의 작업을 정밀하게 예약할 수 있는 고급 스케줄링 프레임워크입니다.

지원하는 스케줄러:

DateTrigger: 특정 시간에 1회 실행
IntervalTrigger: 일정 간격마다 반복
CronTrigger: cron 스타일 스케줄

✔️ 설치:

pip install apscheduler

2. 기본 사용법: 간단한 반복 작업


from apscheduler.schedulers.blocking import BlockingScheduler

def job():
    print("3초마다 실행되는 작업입니다.")

sched = BlockingScheduler()
sched.add_job(job, 'interval', seconds=3)
sched.start()

BlockingScheduler는 프로그램이 종료되지 않도록 루프를 유지합니다.

3. 다양한 트리거 사용법

📌 특정 시간에 1회 실행 (DateTrigger)


from datetime import datetime

sched.add_job(job, 'date', run_date=datetime(2025, 5, 20, 10, 0))

📌 매 5초마다 반복 실행 (IntervalTrigger)

sched.add_job(job, 'interval', seconds=5)

📌 매일 오전 9시에 실행 (CronTrigger)

sched.add_job(job, 'cron', hour=9, minute=0)

✔️ cron 표현식과 유사하게 요일, 시간, 날짜 지정이 가능합니다.

4. 멀티 태스크 처리

APScheduler는 작업을 비동기 혹은 병렬로 실행할 수 있습니다. 여러 작업을 동시에 실행해도 각 작업이 서로 영향을 주지 않도록 스케줄링합니다.


def task1():
    print("작업 1")

def task2():
    print("작업 2")

sched.add_job(task1, 'interval', seconds=5)
sched.add_job(task2, 'interval', seconds=7)
sched.start()

여러 개의 작업이 각자 다른 간격으로 실행되어도 문제없이 돌아갑니다.

5. 작업 상태 확인 및 관리

등록된 작업들을 확인할 수 있습니다.


for job in sched.get_jobs():
    print(job)

또한 특정 작업 ID를 통해 중단하거나 제거할 수 있습니다.


sched.remove_job('my_job_id')

작업 등록 시 ID 지정:


sched.add_job(job, 'interval', seconds=10, id='my_job_id')

6. DB 트리거 기반 자동화 (Event Driven)

예를 들어, DB에 특정 레코드가 들어오면 작업을 실행하는 자동화도 가능합니다. 아래는 SQLite와 연동한 예시입니다.


import sqlite3

def check_db():
    conn = sqlite3.connect("data.db")
    cur = conn.cursor()
    cur.execute("SELECT COUNT(*) FROM alerts WHERE status='new'")
    count = cur.fetchone()[0]
    if count > 0:
        print(f"{count}건의 새로운 알림 감지됨!")
    conn.close()

sched.add_job(check_db, 'interval', seconds=30)
sched.start()

이 방식은 DB 상태를 주기적으로 확인해 자동으로 반응하는 자동화 루틴을 만들 수 있습니다.

7. 백그라운드 실행 (BackgroundScheduler)

BlockingScheduler는 코드 실행을 멈추기 때문에, 일반 파이썬 앱에서 사용하기엔 불편할 수 있습니다. 그럴 땐 BackgroundScheduler를 사용하세요.


from apscheduler.schedulers.background import BackgroundScheduler

sched = BackgroundScheduler()
sched.add_job(job, 'interval', seconds=10)
sched.start()

# 아래는 메인 애플리케이션 코드
while True:
    print("메인 앱 실행 중...")
    time.sleep(5)

이렇게 하면 다른 코드와 함께 APScheduler를 백그라운드로 돌릴 수 있습니다.

8. 로그 남기기 및 에러 핸들링

스케줄 실행 상태나 에러를 기록하려면 logging을 활용하세요.


import logging

logging.basicConfig()
logging.getLogger('apscheduler').setLevel(logging.DEBUG)

또는 예외 처리 콜백을 등록할 수도 있습니다.


def listener(event):
    print("에러 발생!", event)

sched.add_listener(listener)

업무 자동화에 강력한 파트너

APScheduler는 단순한 반복 작업을 넘어 정밀한 스케줄 관리, 멀티 태스크 처리, DB 감시, 유지보수 가능한 루틴 구성까지 지원합니다. schedule 모듈이 입문자용이라면, APScheduler는 실무용 자동화 프레임워크라 할 수 있습니다.

특히 아래와 같은 상황에서 APScheduler가 빛을 발합니다:

매일 여러 시간대에 다양한 업무를 처리해야 할 때
DB 상태나 외부 이벤트에 따라 반응해야 할 때
여러 작업을 동시에 안정적으로 실행하고 관리해야 할 때

파이썬 자동화의 끝판왕을 찾고 있다면, APScheduler는 반드시 익혀야 할 도구입니다. 오늘 하나의 작업부터 시작해보세요!

파이나라