[Django Apscheduler] django에 스케줄러 도입해 batch 만들기

Batch 서버는 Job(실행 단위 정도..?)을 관리할 뿐 엄밀하게 스케줄러는 아닙니다. 하지만 보통 스케줄러와 함께 정기적으로 로직을 실행한다는 의미로 이 글에서 표현했습니다.

🔹 라이브러리 선정

Batch를 구현하기 위해서는 외부에서 정기적으로 트리거를 발생시키거나 django 내부에서 정기적으로 로직을 실행해야 합니다. 리눅스 환경에서 django 자체적으로 os 지식 없이 batch를 구현하는 라이브러리를 선정했습니다. 

 

후보 1. django-crontab

글쓰는 당시 django batch를 검색했을때 가장 많이 나왔습니다. cron은 유닉스 계열 운영 체제의 시간 기반 잡 스케줄러입니다. 최신글이 계속 나오는것으로 보아 아직까지 호환되는 라이브러리 같습니다. 하지만, 2016년에 마지막 업데이트를 이후로 더이상 라이브러리 업데이트가 진행되고 있지 않습니다(링크). 따라서, 이후 취약점이나 버전 호환성에 문제가 있어 선택하지 않았습니다.

 

후보 2. django-apscheduler

django에 스케줄러를 달아 정기적으로 로직을 실행할 수 있게 하는 라이브러리입니다. Batch 서버가 요구하는 트랜잭션, 작업 재시작, 중복 작업 건너뛰기 등 충분한 기능을 지원하지는 않습니다. (본인도 구현 못하면서 Spring batch를 쓰다와서 눈만 높아짐)

로깅이나 실행 인스턴스 수 설정 등 기본적으로 배치로 쓸수있는 기능을 지원하고, 나머지는 로직에서 구현하면 되어 이 라이브러리를 선택합니다. (링크)

 

🔹 적용하기

1. 라이브러리 설치

pip install django-apscheduler

 

2. settings 등록

INSTALLED_APPS = (
    # ...
    "django_apscheduler",
)

APSCHEDULER_DATETIME_FORMAT = "N j, Y, f:s a"

APSCHEDULER_RUN_NOW_TIMEOUT = 25  # Seconds

SCHEDULER_DEFAULT = True # apps.py 참고

• Installed_apps는 django에 해당 라이브러리를 인식시킵니다.
• 그 아래 format은 django 관리 사이트에서 런타임 타임스탬프를 표시하기 위한 형식 문자열입니다.
  구문 세부 정보는 https://docs.djangoproject.com/en/dev/ref/settings/#datetime-format을 참조하세요.
  format이 없어도 서비스는 동작합니다.
• APSCHEDULER_RUN_NOW_TIMEOUT은 타임아웃(최대 실행 시간) 설정값이고, 디폴트가 25초 입니다. 
  작업 시간에 따라 긴 로직을 수행하는 프로젝트에서는 시간을 길게 설정해야 합니다.

 

3. Job 등록

# runapscheduler.py
import logging

from apscheduler.schedulers.background import BackgroundScheduler
from django_apscheduler.jobstores import register_events, DjangoJobStore

logger = logging.getLogger(__name__)


def my_job_a():
  # 실행시킬 Job
  # 여기서 정의하지 않고, import 해도 됨
  pass
  
def my_job_b():
  # 실행시킬 Job
  # 여기서 정의하지 않고, import 해도 됨
  pass

def start():
  def handle(self, *args, **options):
    scheduler = BackgroundScheduler(timezone=settings.TIME_ZONE) # BlockingScheduler를 사용할 수도 있습니다.
    scheduler.add_jobstore(DjangoJobStore(), "default") 

    scheduler.add_job(
      my_job_a,
      trigger=CronTrigger(second="*/60"),  # 60초마다 작동합니다.
      id="my_job",  # id는 고유해야합니다. 
      max_instances=1,
      replace_existing=True,
    )
    logger.info("Added job 'my_job_a'.")

    scheduler.add_job(
      my_job_b,
      trigger=CronTrigger(
        day_of_week="mon", hour="03", minute="00"
      ),  # 실행 시간입니다. 여기선 매주 월요일 3시에 실행합니다.
      id="my_job_b",
      max_instances=1,
      replace_existing=True,
    )
	logger.info("Added job 'my_job_b'.")

    try:
      logger.info("Starting scheduler...")
      scheduler.start() # 없으면 동작하지 않습니다.
    except KeyboardInterrupt:
      logger.info("Stopping scheduler...")
      scheduler.shutdown()
      logger.info("Scheduler shut down successfully!")

• 해당 코드 설명은 주석으로 달았습니다.

 

4. app.py에 등록

# apps.py

from django.apps import AppConfig
from django.conf import settings


class MyAppConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'myapp'

    def ready(self):
        if settings.SCHEDULER_DEFAULT:
            from . import runapscheduler
            runapscheduler.start()

 

🔹 [참고] 복합적 cronjob 시간 설정 예시 (trigger) 

from apscheduler.triggers.combining import OrTrigger
from apscheduler.triggers.cron import CronTrigger

trigger = OrTrigger(
    CronTrigger(day_of_week="mon-fri", hour=10),
    CronTrigger(day_of_week="sat-sun", hour=11),
)

from apscheduler.triggers.calendarinterval import CalendarIntervalTrigger
from apscheduler.triggers.combining import AndTrigger
from apscheduler.triggers.cron import CronTrigger

trigger = AndTrigger(
    CalendarIntervalTrigger(months=2, hour=10),
    CronTrigger(day_of_week="mon-fri", hour=10),
)

 

* 공식 문서

https://pypi.org/project/django-apscheduler/