Apscheduler
Apscheduler란?
- Python code에 대한 Scheduling을 구성해주는 library
Template code
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
from apscheduler.jobstores.base import JobLookupError
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.schedulers.background import BackgroundScheduler
from .logger import Logger
import time
import datetime
logger = Logger(method='scheduler', dir='log', filename='scheduler.log')
class Scheduler(object):
def __init__(self):
logger.set_log('INFO', "Scheduler is initialized", extra={'classAndFunc': self.__init__.__qualname__})
self.sched = BackgroundScheduler({
'apscheduler.timezone': 'Asia/Seoul',
'apscheduler.job_defaults.max_instances': '7'
})
self.sched.start()
self.job_id = ''
def __del__(self):
self.shutdown()
def shutdown(self):
logger.set_log('DEBUG', "Shutdown", extra={'classAndFunc': self.shutdown.__qualname__})
self.sched.shutdown()
def kill_scheduler(self, job_id):
try:
self.sched.remove_job(job_id)
except JobLookupError as e:
logger.set_log('INFO', "Failed to stop Scheduler : {0}".format(e), extra={'classAndFunc': self.kill_scheduler.__qualname__})
return
def run_job(self, type, job_id, function, sleep_time=0, *args, **kwargs):
if sleep_time != 0:
time.sleep(sleep_time)
try:
result = function(*args, **kwargs)
logger.set_log('DEBUG', "{0} Scheduler p_id[{1}] : {2} // Sleep time : {3}".format(type, job_id, time.localtime().tm_sec, sleep_time), extra={'classAndFunc': self.run_job.__qualname__})
except Exception as e:
result = None
logger.set_log('INFO', "Failed to run function : {0}".format(e), extra={'classAndFunc': self.run_job.__qualname__})
def scheduler(self, type: str, job_id: str, sleep_time=0, function=None, seconds: int=1, year: str=None, month: str=None, day: str=None, week: str=None, day_of_week: str='mon-fri', hour: str=None, minute: str=None, second: str=None, *args, **kwargs):
logger.set_log('INFO', "{0} Scheduler start".format(type), extra={'classAndFunc': self.scheduler.__qualname__})
if type == 'interval':
self.sched.add_job(self.run_job, type, seconds=seconds, id=job_id, args=(type, job_id, function, sleep_time) + args, kwargs=kwargs)
elif type == 'cron':
self.sched.add_job(self.run_job, type,
year=year, month=month, day=day, week=week,
day_of_week=day_of_week,
hour=hour, minute=minute, second=second,
id=job_id, args=(type, job_id, function, sleep_time) + args, kwargs=kwargs)
#elif type == 'cron_current':
### self.sched.add_job(self.run_job, type, day_of_week='mon-fri', hour='9-15', second='*/5', id=job_id, args=(type, job_id, code, postfix, sleep_time))
if __name__=="__main__":
def hello(name):
print(name)
print("Hello!")
scheduler = Scheduler()
scheduler.scheduler(a="Taelim", type='cron', job_id='1', second='*/1', sleep_time=0, function=hello)
while True:
logger.set_log('INFO', "{0} Scheduler is running...".format(type), extra={'classAndFunc': 'main'})
time.sleep(600)
*args, **kwargs를 넘기려면?
1
2
3
4
5
6
### add_job method에 parameter 넘길 시, args는 tuple 합연산으로, kwargs는 kwargs 변수로 넘기면(또는 다른 kwargs 변수와 합친 상태로) 된다
self.sched.add_job(self.run_job, type,
year=year, month=month, day=day, week=week,
day_of_week=day_of_week,
hour=hour, minute=minute, second=second,
id=job_id, args=(type, job_id, function, sleep_time) + args, kwargs=kwargs)