added cronjob decorator code

This commit is contained in:
brb
2023-07-24 09:37:41 +02:00
parent d13e069287
commit be69763f48
+53
View File
@@ -0,0 +1,53 @@
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')