APScheduler 스케쥴링을 동적으로 제어하는 방법

APScheduler 의 스케쥴링을 동적으로 제어하기 위해 사용한 방법을 간단히 정리해보자.

hydra(A framework for elegantly configuring complex applications)를 사용 중인데 이 프레임워크는 OmegaConf 로 설정값을 관리한다. 이 설정값으로 APScheduler의 스케쥴링을 동적으로 제어하고자 한다.

간단하므로 코드 부터 우선.

#config.yaml
...

queries:
  test0:
     trigger: interval
     time:
        seconds: 5

  test1:
     trigger: cron
     time:
         second: '*/5'

  test2: 
     trigger: cron 
     time: 
         minute: '*'
         second: '5'

def start_scheduler(cfg):
    scheduler = BackgroundScheduler()

    for key, value in cfg.queries.items():
        scheduler.add_job(exec_processing, value['trigger'], args=[cfg, key], **value['time'])

    scheduler.start()
   
    try:
        while True:
            time.sleep(2)
    except (KeyboardInterrupt, SystemExit):
        scheduler.shutdown()


def exec_processing(cfg, key):
    pring(cfg.queries[key])


@hydra.main(config_name="config")
def my_app(cfg: DictConfig) -> None:
    start_scheduler(cfg)


if __name__ == "__main__":
    my_app()

APScheduler add_job 메서드의 패러미터를 보면 첫 번째가 호출할 함수명, 두 번째가 trigger 의 종류(cron, interval, date), args 는 호출 함수에 넘겨줄 패러미터를 지정해주도록 되어있다. 그리고, 이번 주제에서 다룰 스케쥴링 주기에 대한 설정값은 **trigger_args 에 dict 타입으로 넘겨주면 된다.

add_job(func, trigger=None, args=None, kwargs=None, id=None, name=None, misfire_grace_time=undefined, coalesce=undefined, max_instances=undefined, next_run_time=undefined, jobstore='default', executor='default', replace_existing=False, **trigger_args)

위 코드의 for 문은 yaml 의 queries 설정값들로 아래처럼 설정을 해주게 될거다.

scheduler.add_job(exec_processing, 'interval', args=[cfg, 'test0'], **{'seconds': 5})  # 5초 간격으로 실행
scheduler.add_job(exec_processing, 'cron', args=[cfg, 'test1'], **{'second': '*/5'})   # 5, 10, 15, ... 에 실행
scheduler.add_job(exec_processing, 'cron', args=[cfg, 'test2'], **{'minute': '*', 'second': '5'})  # 매분 5초에 실행

**trigger_args 다루는 것 같은 기초 수준 문제에서 헛발질을 꽤 했다. 주위에 자바 개발자 밖에 없어 물어볼 곳 마땅찮으니 파이썬 초보 넘어서기가 이리 더디다.

apscheduler.schedulers.base — APScheduler 3.6.3 documentation

Hydra | Hydra