54 lines
1.5 KiB
Python
54 lines
1.5 KiB
Python
from cron_converter import Cron
|
|
from configparser import ConfigParser
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
import dateutil
|
|
import time
|
|
import functools
|
|
import logging
|
|
|
|
# load default values
|
|
config = ConfigParser()
|
|
config.read(Path(__file__).parent / 'defaults.ini')
|
|
SCHEDULE = config.get('cron', 'schedule')
|
|
TZ = config.get('cron', 'tz')
|
|
|
|
|
|
def cronjob(
|
|
schedule: str = SCHEDULE,
|
|
tz: str = TZ
|
|
):
|
|
"""
|
|
decorator for running function as a cronjob.
|
|
"""
|
|
logging.info(f'started cronjob with schedule {schedule} and tz {tz}')
|
|
|
|
def decorator_cronjob(func):
|
|
@functools.wraps(func)
|
|
def wrapper_cronjob(*args, **kwargs):
|
|
scheduler = Cron(schedule).schedule(timezone_str=tz)
|
|
trigger_time = scheduler.next()
|
|
while True:
|
|
while trigger_time > datetime.now(tz=dateutil.tz.gettz(tz)):
|
|
# waiting loop
|
|
time.sleep(1)
|
|
# actual execution
|
|
try:
|
|
func(*args, **kwargs)
|
|
except Exception as e:
|
|
logging.warning(f'failed executing {func}: {e}')
|
|
raise
|
|
else:
|
|
# update trigger time
|
|
trigger_time = scheduler.next()
|
|
return wrapper_cronjob
|
|
return decorator_cronjob
|
|
|
|
|
|
if __name__ == '__main__':
|
|
@cronjob(schedule='* * * * *', tz=TZ)
|
|
def test_loop(msg: str):
|
|
print(msg)
|
|
|
|
test_loop('hello world')
|