[Python] APScheduler 가이드

📌1. APScheduler 설치하기

pip install apscheduler

 

추가적인 의존성 설치

• APScheduler는 기본적으로 `threading`을 사용하지만, 실행 환경에 따라 다른 방식도 지원합니다.
• 특정 기능을 사용하려면 아래 패키지도 함께 설치해야 할 수 있습니다.

pip install apscheduler[asyncio] # asyncio 환경 지원 pip install apscheduler[gevent] # gevent 환경 지원 pip install apscheduler[tornado] # tornado 환경 지원 pip install apscheduler[redis] # Redis 기반 Job Store 사용

📌 2. APScheduler의 주요 개념

APScheduler는 여러 개의 작업(Job)을 예약하여 실행하는 기능을 제공합니다.

이를 위해 아래와 같은 개념을 이해해야 합니다.

개념	설명
Job Store (작업 저장소)	예약된 작업을 저장하는 공간 (기본적으로 메모리에 저장되지만, DB, Redis 등도 가능)
Trigger (트리거)	작업이 실행될 조건을 정의 (특정 시간, 주기적으로 실행 등)
Executor (실행기)	작업을 실행하는 방식 (스레드, 프로세스, 비동기 등)
Scheduler (스케줄러)	전체 작업을 관리하고 실행하는 역할

 

---

📌 3. APScheduler 사용 예제

1️⃣ 기본 스케줄러 실행

from apscheduler.schedulers.background import BackgroundScheduler
import time

def my_task():
	print("작업이 실행되었습니다!")
    
scheduler = BackgroundScheduler()
scheduler.add_job(my_task, 'interval', seconds=10) # 10초마다 실행
scheduler.start()

try:
	while True:
    	time.sleep(1) # 프로그램이 종료되지 않도록 유지
except KeyboardInterrupt:
	scheduler.shutdown()

✅ 설명:

• `BackgroundScheduler()`를 사용하여 스케줄러를 실행
• `add_job()`을 사용해 10초마다 실행되는 작업을 추가
• `scheduler.start()`로 실행 시작

---

2️⃣ 특정 시간에 한 번 실행

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime

def my_task():
	print(f"작업 실행됨: {datetime.now()}")

scheduler = BlockingScheduler()
scheduler.add_job(my_task, 'date', run_date='2025-02-20 15:30:00')
scheduler.start()

 

✅ 설명:

• `date` 트리거를 사용해 특정 시간(2025년 2월 20일 15시 30분)에 실행
• `BlockingScheduler()`는 프로그램이 실행된 채로 유지됨

---

3️⃣ 매일 특정 시간에 실행

from apscheduler.schedulers.background import BackgroundScheduler

scheduler = BackgroundScheduler()

scheduler.add_job(my_task, 'cron', hour=0, minute=30) # 매일 오전 9시 30분 실행
scheduler.start()

✅ 설명:

• `cron` 트리거를 사용해 매일 특정 시간 (9시 30분) 에 실행

---

4️⃣ 매주 월요일, 수요일, 금요일 실행

scheduler.add_job(my_task, 'cron', day_of_week = 'mon,wed,fri', hour=10, minute=0)

✅ 설명:

• `day_of_week` 를 사용해 매주 월/수/금 오전 10시에 실행

---

5️⃣ 동적으로 작업 추가 및 제거

job = scheduler.add_job(my_task, 'interval', minute=5)
job.remove() # 작업 삭제

✅ 설명:

• `add_job` 을 통해 동적으로 작업 추가
• `remove()`를 호출하여 특정 작업 삭제 가능

---

6️⃣ 실행 중인 모든 작업 조회

for job in scheduler.get_jobs():
	print(f"등록된 작업: {job}")

✅ 설명:

• `get_jobs()`를 호출하면 현재 등록된 모든 작업 목록을 가져올 수 있음

---

📌 4. APScheduler에서 Job Store 사용하기

기본적으로 APScheduler는 메모리에 작업을 저장하지만, 파일 또는 데이터베이스에 저장할 수도 있음.

1️⃣ SQLite에 작업 저장

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore

jobstores = {
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}

scheduler = BackgroundScheduler(jobstores=jobstores)
scheduler.start()

 

✅ 설명:

• APScheduler의 Job Store를 SQLite로 설정하면, 작업이 데이터베이스에 저장됨
• `sqlite:///jobs.sqlite`는 현재 폴더에 jobs.sqlite 파일을 생성함

2️⃣ Redis에 작업 저장

from apscheduler.jobstores.redis import RedisJobStore

jobstores = {
    'default': RedisJobStore(host='localhost', port=6379)
}

scheduler = BackgroundScheduler(jobstores=jobstores)
scheduler.start()

✅ 설명:

• Redis를 사용하여 Job Store를 설정하면, 작업이 Redis에 저장됨
• 이 방법은 여러 서버에서 동일한 작업을 공유할 때 유용함

---

📌 5. APScheduler에서 Executor 설정하기

Executor는 작업을 실행하는 방식을 결정함

1️⃣ 기본 설정 (ThreadPoolExecutor)

from apscheduler.executors.pool import ThreadPoolExecutor

executors = {
    'default': ThreadPoolExecutor(max_workers=5)  # 최대 5개의 작업 동시 실행
}

scheduler = BackgroundScheduler(executors=executors)
scheduler.start()

✅ 설명:

• ThreadPoolExecutor(max_workers=5)를 사용하면 최대 5개의 작업을 동시에 실행할 수 있음

2️⃣ 멀티 프로세스 실행 (ProcessPoolExecutor)

from apscheduler.executors.pool import ProcessPoolExecutor

executors = {
    'default': ProcessPoolExecutor(max_workers=3)  # 멀티 프로세스로 실행
}

scheduler = BackgroundScheduler(executors=executors)
scheduler.start()

✅ 설명:

• ProcessPoolExecutor를 사용하면 멀티 프로세스를 활용하여 실행 가능
• CPU 부하가 높은 작업에 적합

---

📌 6. APScheduler의 실행 방식

APScheduler는 4가지 방식으로 실행할 수 있음.

실행 방식		설명
BlockingScheduler		프로그램이 종료되지 않고 유지됨 (단일 프로세스에서 실행할 때 사용)
BackgroundScheduler		백그라운드에서 실행됨 (일반적인 애플리케이션에 적합)
AsyncIOScheduler		asyncio 이벤트 루프에서 실행됨 (비동기 작업에 적합)
TornadoScheduler		Tornado 프레임워크에서 실행됨

 

✅ 일반적인 사용:

scheduler = BackgroundScheduler()
scheduler.start()

 

✅ FastAPI 또는 Flask 같은 웹 서버에서는:

scheduler = BackgroundScheduler()
scheduler.start(paused=True)  # 웹 애플리케이션과 함께 실행

---

📌 7. 결론

APScheduler를 사용하면 스케줄링이 필요한 작업을 간편하게 자동화할 수 있습니다.

• `interval`, `date`, `cron` 트리거를 활용하여 다양한 실행 주기를 설정 가능
• SQLite, Redis 같은 Job Store를 이용하여 작업을 저장 가능
• Thread, Process, Async 등을 활용해 실행 방식 조정 가능