APScheduler (작업 스케줄러 파이썬 라이브러리)

00 개요

• 코드분석 중 서비스를 실행시킬 때 APScheduler라는 라이브러리를 사용하는데 이에 대해 정리하고자 함

01 APScheduler란

1. 정의

• 특정 시간에 작업을 실행하거나, 주기적으로 작업을 예약하는 기능 제공
• 다양한 작업 예약 방법을 지원
• 웹 애플리케이션, 데이터 수집, 배치 작업 등 여러 용도로 사용됨

2. 기본 개념 및 주요 기능

• APScheduler의 4가지 요소: triggers (트리거), job stores (작업 저장소), executors (실행자), schedulers (스케줄러)

1) 다양한 Trigger 제공

• Trigger란
  • Triggers contain the scheduling logic.
  • Each job has its own trigger which determines when the job should be run next.
  • Beyond their initial configuration, triggers are completely stateless.
• Trigger 종류
  • Date Trigger: 특정 날짜와 시간에 한 번 작업 실행
  • Interval Trigger: 일정 간격으로 작업 반복 실행
  • Cron Trigger: 크론 표현식을 사용하여 복잡한 일정으로 작업 예약

2) 유연한 Job Stores 

• Job store란
  • Job store houses the scheduled jobs.
  • Job store keeps the jobs in memory (기본값), or in e.g., Redis, or in various kinds of DBs(e.g., SQLite, PostgreSQL 등)
  • A job’s data is serialized when it is saved to a persistent job store, and deserialized when it’s loaded back from it.
  • Job stores (other than the default one) don’t keep the job data in memory, but act as middlemen for saving, loading, updating and searching jobs in the backend.
  • Job stores must never be shared between schedulers.

3) Executors - 이벤트 시스템 및 백그라운드 실행

• Executors란
  • Executors are what handle the running of the jobs.
  • They do this typically by submitting the designated callable in a job to a thread or process pool.
  • When the job is done, the executor notifies the scheduler which then emits an appropriate event.

• 이벤트 시스템 - 작업 실행, 완료, 오류 등 이벤트를 트리거하고, 이를 처리할 listerner 등록 가능
• 백그라운드 실행 - 메인 프로그램의 실행을 방해하지 않고 백그라운드에서 작업 실행 가능

5) 다양한 Schedulers 

• Scheduler란
  • Schedulers are what bind the rest together.
  • You typically have only one scheduler running in your application.
  • The application developer doesn’t normally deal with the job stores, executors or triggers directly.
  • Instead, the scheduler provides the proper interface to handle all those.
  • Configuring the job stores and executors is done through the scheduler, as is adding, modifying and removing jobs.
• Scheduler 종류
  • BlockingScheduler: use when the scheduler is the only thing running in your process 
  • BackgroundScheduler: use when you’re not using any of the frameworks below, and want the scheduler to run in the background inside your application 
  • AsyncIOScheduler: use if your application uses the asyncio module 
  • GeventScheduler: use if your application uses gevent 
  • TornadoScheduler: use if you’re building a Tornado application 
  • TwistedScheduler: use if you’re building a Twisted application 
  • QtScheduler: use if you’re building a Qt application 

3. 알맞은 Scheduler, Job store(s), Executor(s) and Trigger(s) 고르기

1) Scheduler 고르기

• BlockingScheduler: use when the scheduler is the only thing running in your process 
• BackgroundScheduler: use when you’re not using any of the frameworks below, and want the scheduler to run in the background inside your application 
• AsyncIOScheduler: use if your application uses the asyncio module 
• GeventScheduler: use if your application uses gevent 
• TornadoScheduler: use if you’re building a Tornado application 
• TwistedScheduler: use if you’re building a Twisted application 
• QtScheduler: use if you’re building a Qt application 

2) Job Store 고르기

• To pick the appropriate job store, you need to determine whether you need job persistence or not.
• If you always recreate your jobs at the start of your application, then you can probably go with the default (MemoryJobStore).
• But if you need your jobs to persist over scheduler restarts or application crashes, then your choice usually boils down to what tools are used in your programming environment.
• If, however, you are in the position to choose freely, then SQLAlchemyJobStore on a PostgreSQL backend is the recommended choice due to its strong data integrity protection.

3) Executor 고르기

• Likewise, the choice of executors is usually made for you if you use one of the frameworks above. 
• Otherwise, the default ThreadPoolExecutor should be good enough for most purposes.
• If your workload involves CPU intensive operations, you should consider using ProcessPoolExecutor instead to make use of multiple CPU cores.
• You could even use both at once, adding the process pool executor as a secondary executor.

4) Trigger 고르기

• When you schedule a job, you need to choose a trigger for it. 
• The trigger determines the logic by which the dates/times are calculated when the job will be run. 
• 3 built-in trigger types:
  • date: use when you want to run the job just once at a certain point of time
  • interval: use when you want to run the job at fixed intervals of time
  • cron: use when you want to run the job periodically at certain time(s) of day
• It is also possible to combine multiple triggers into one which fires either on times agreed on by all the participating triggers, or when any of the triggers would fire.

4. Scheduler events

• It is possible to attach event listeners to the scheduler.
• Scheduler events are fired on certain occasions, and may carry additional information in them concerning the details of that particular event.
• It is possible to listen to only particular types of events by giving the appropriate mask argument to add_listener(), OR’ing the different constants together.
• The listener callable is called with one argument, the event object. (참조: events and their attributes)

0) 예시 - listener 추가하기

def my_listener(event):
    if event.exception:
        print('The job crashed :(')
    else:
        print('The job worked :)')

scheduler.add_listener(my_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)

1) apscheduler.events 모듈

• 참조: https://github.com/agronholm/apscheduler/blob/3.x/apscheduler/events.py

__all__ = ('EVENT_SCHEDULER_STARTED', 'EVENT_SCHEDULER_SHUTDOWN', 'EVENT_SCHEDULER_PAUSED',
           'EVENT_SCHEDULER_RESUMED', 'EVENT_EXECUTOR_ADDED', 'EVENT_EXECUTOR_REMOVED',
           'EVENT_JOBSTORE_ADDED', 'EVENT_JOBSTORE_REMOVED', 'EVENT_ALL_JOBS_REMOVED',
           'EVENT_JOB_ADDED', 'EVENT_JOB_REMOVED', 'EVENT_JOB_MODIFIED', 'EVENT_JOB_EXECUTED',
           'EVENT_JOB_ERROR', 'EVENT_JOB_MISSED', 'EVENT_JOB_SUBMITTED', 'EVENT_JOB_MAX_INSTANCES',
           'SchedulerEvent', 'JobEvent', 'JobExecutionEvent', 'JobSubmissionEvent')


EVENT_SCHEDULER_STARTED = EVENT_SCHEDULER_START = 2 ** 0
EVENT_SCHEDULER_SHUTDOWN = 2 ** 1
EVENT_SCHEDULER_PAUSED = 2 ** 2
EVENT_SCHEDULER_RESUMED = 2 ** 3
EVENT_EXECUTOR_ADDED = 2 ** 4
EVENT_EXECUTOR_REMOVED = 2 ** 5
EVENT_JOBSTORE_ADDED = 2 ** 6
EVENT_JOBSTORE_REMOVED = 2 ** 7
EVENT_ALL_JOBS_REMOVED = 2 ** 8
EVENT_JOB_ADDED = 2 ** 9
EVENT_JOB_REMOVED = 2 ** 10
EVENT_JOB_MODIFIED = 2 ** 11
EVENT_JOB_EXECUTED = 2 ** 12
EVENT_JOB_ERROR = 2 ** 13
EVENT_JOB_MISSED = 2 ** 14
EVENT_JOB_SUBMITTED = 2 ** 15
EVENT_JOB_MAX_INSTANCES = 2 ** 16
EVENT_ALL = (EVENT_SCHEDULER_STARTED | EVENT_SCHEDULER_SHUTDOWN | EVENT_SCHEDULER_PAUSED |
             EVENT_SCHEDULER_RESUMED | EVENT_EXECUTOR_ADDED | EVENT_EXECUTOR_REMOVED |
             EVENT_JOBSTORE_ADDED | EVENT_JOBSTORE_REMOVED | EVENT_ALL_JOBS_REMOVED |
             EVENT_JOB_ADDED | EVENT_JOB_REMOVED | EVENT_JOB_MODIFIED | EVENT_JOB_EXECUTED |
             EVENT_JOB_ERROR | EVENT_JOB_MISSED | EVENT_JOB_SUBMITTED | EVENT_JOB_MAX_INSTANCES)


class SchedulerEvent(object):
    """
    An event that concerns the scheduler itself.

    :ivar code: the type code of this event
    :ivar alias: alias of the job store or executor that was added or removed (if applicable)
    """

    def __init__(self, code, alias=None):
        super(SchedulerEvent, self).__init__()
        self.code = code
        self.alias = alias

    def __repr__(self):
        return '<%s (code=%d)>' % (self.__class__.__name__, self.code)


class JobEvent(SchedulerEvent):
    """
    An event that concerns a job.

    :ivar code: the type code of this event
    :ivar job_id: identifier of the job in question
    :ivar jobstore: alias of the job store containing the job in question
    """

    def __init__(self, code, job_id, jobstore):
        super(JobEvent, self).__init__(code)
        self.code = code
        self.job_id = job_id
        self.jobstore = jobstore


class JobSubmissionEvent(JobEvent):
    """
    An event that concerns the submission of a job to its executor.

    :ivar scheduled_run_times: a list of datetimes when the job was intended to run
    """

    def __init__(self, code, job_id, jobstore, scheduled_run_times):
        super(JobSubmissionEvent, self).__init__(code, job_id, jobstore)
        self.scheduled_run_times = scheduled_run_times


class JobExecutionEvent(JobEvent):
    """
    An event that concerns the running of a job within its executor.

    :ivar scheduled_run_time: the time when the job was scheduled to be run
    :ivar retval: the return value of the successfully executed job
    :ivar exception: the exception raised by the job
    :ivar traceback: a formatted traceback for the exception
    """

    def __init__(self, code, job_id, jobstore, scheduled_run_time, retval=None, exception=None,
                 traceback=None):
        super(JobExecutionEvent, self).__init__(code, job_id, jobstore)
        self.scheduled_run_time = scheduled_run_time
        self.retval = retval
        self.exception = exception
        self.traceback = traceback

2) apscheduler.events.py 모듈 정리

API

class	설명	variables
apscheduler.events.SchedulerEvent(code, alias=None)	An event that concerns the scheduler itself.	code – the type code of this event alias – alias of the job store or executor that was added or removed (if applicable)
apscheduler.events.JobEvent(code, job_id, jobstore)	Bases: SchedulerEvent An event that concerns a job.	code – the type code of this event job_id – identifier of the job in question jobstore – alias of the job store containing the job in question
apscheduler.events.JobSubmissionEvent(code, job_id, jobstore, scheduled_run_times)	Bases: JobEvent An event that concerns the submission of a job to its executor.	scheduled_run_times – a list of datetimes when the job was intended to run
apscheduler.events.JobExecutionEvent(code, job_id, jobstore, scheduled_run_time, retval=None, exception=None, traceback=None)	Bases: JobEvent An event that concerns the running of a job within its executor.	scheduled_run_time – the time when the job was scheduled to be run retval – the return value of the successfully executed job exception – the exception raised by the job traceback – a formatted traceback for the exception

Event Codes

• event codes are numeric constants importable from apscheduler.events

Constant	설명	Event Class
EVENT_SCHEDULER_STARTED	The scheduler was started	SchedulerEvent
EVENT_SCHEDULER_SHUTDOWN	The scheduler was shut down	SchedulerEvent
EVENT_SCHEDULER_PAUSED	Job processing in the scheduler was paused	SchedulerEvent
EVENT_SCHEDULER_RESUMED	Job processing in the scheduler was resumed	SchedulerEvent
EVENT_EXECUTOR_ADDED	An executor was added to the scheduler	SchedulerEvent
EVENT_EXECUTOR_REMOVED	An executor was removed to the scheduler	SchedulerEvent
EVENT_JOBSTORE_ADDED	A job store was added to the scheduler	SchedulerEvent
EVENT_JOBSTORE_REMOVED	A job store was removed from the scheduler	SchedulerEvent
EVENT_ALL_JOBS_REMOVED	All jobs were removed from either all job stores or one particular job store	SchedulerEvent
EVENT_JOB_ADDED	A job was added to a job store	JobEvent
EVENT_JOB_REMOVED	A job was removed from a job store	JobEvent
EVENT_JOB_MODIFIED	A job was modified from outside the scheduler	JobEvent
EVENT_JOB_SUBMITTED	A job was submitted to its executor to be run	JobSubmissionEvent
EVENT_JOB_MAX_INSTANCES	A job being submitted to its executor was not accepted by the executor because the job has already reached its maximum concurrently executing instances	JobSubmissionEvent
EVENT_JOB_EXECUTED	A job was executed successfully	JobExecutionEvent
EVENT_JOB_ERROR	A job raised an exception during execution	JobExecutionEvent
EVENT_JOB_MISSED	A job’s execution was missed	JobExecutionEvent
EVENT_ALL	A catch-all mask that includes every event type	N/A

 

 

 

 

<예시>

scheduler = BackgroundScheduler(job_defaults={ 
                                        'coalesce': True, 
                                        'max_instances': 30, 
                                        'misfire_grace_time': 1800 })
scheduler.start()
scheduler.add_job(self.polling_job, 
                    "interval", 
                    seconds=time, 
                    name=job_id, 
                    id=job_id, 
                    args=[job_id, time], 
                    next_run_time=run_time , 
                    replace_existing=True)
scheduler.get_jobs()

def listener(self, event): # APScheduler event handler
    if event.exception:
        self.logger.debug(f'loaderMain , event error {event}')

scheduler.add_listener(self.listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)

• 마지막 줄 설명:
  • scheduler:
    • APScheduler의 스케줄러 인스턴스
    • 이 객체는 예약된 작업(job)을 관리 및 실행
  • 스케줄러.add_listener():
    • APScheduler의 메서드
    • 특정 이벤트가 발생할 때 호출할 listener(콜백 함수)를 등록
  • self.listener:
    • 이벤트가 발생했을 때 호출될 위에 정의된 함수
    • 이 함수는 특정 작업에 오류가 발생했을 때 처리할 로직 (logger의 debug 레벨에 'loaderMain, event error <이벤트이름>' 메세지 기록) 포함
  • EVENT_JOB_EXECUTED: 
    • 작업이 정상적으로 완료되었을 때 발생하는 이벤트
    • APScheduler는 작업이 성공적으로 끝날 때 이 이벤트를 trigger함
  • EVENT_JOB_ERROR: 
    • 작업이 실행되는 동안 오류가 발생했을 때 발생하는 이벤트
    • APScheduler는 작업이 실패했을 때 이 이벤트를  trigger함
  • | (비트 OR 연산자):
    • 두 이벤트를 결합하는 역할
    • 이 연산자는 EVENT_JOB_EXECUTED 또는 EVENT_JOB_ERROR 이벤트 중 하나가 발생할 때 리스너가 호출되도록 함
    1.  
• 따라서, 이 코드는 스케줄러에서 작업이 성공적으로 완료되거나 오류가 발생할 때마다 self.listener 함수를 호출하도록 설정

 

 

참조

1. https://apscheduler.readthedocs.io/en/3.x/userguide.html 
2. (apscheduler events 모듈) https://github.com/agronholm/apscheduler/blob/3.x/apscheduler/events.py
3.  
4.  
5.  
6.