periodic_test.py

#!/usr/bin/env python
# using https://github.com/escapewindow/python-generic-worker/blob/1046f6d177554dc3ddf2c14110849426913df00b/test.py
#  linked to lib.py
import asyncio
import atexit
import logging
import time

#log = logging.getLogger(__name__)
from lib import update_logging_config, log_errors, read_stdout, get_log_filenames, get_log_fhs, run_task, Context, close_asyncio_loop, log

def get_date_string():
    return time.strftime("%H:%M:%S", time.localtime())


def periodic(context, period=5):
    if context.status not in ("done", None):
        log.debug("Periodic running")
        loop = asyncio.get_event_loop()
        loop.call_later(period, periodic, context)
    else:
        log.debug("Peridic not running")


def main():
    context = Context()
    context.config = {
        'log_dir': "/tmp",
        'verbose': True,
        'task_script': [
            'bash', '-c',
            r'''echo {} && (>&2 echo "error") && sleep 19 && exit 2'''.format(
                get_date_string()
            ),
        ],
    }
    update_logging_config(context, log)
    log.debug("DEBUG")
    loop = asyncio.get_event_loop()
    atexit.register(close_asyncio_loop)
    context.status = "running"
    loop.call_later(5, periodic, context)
    loop.run_until_complete(run_task(context))
    context.status = "done"
    log.debug(asyncio.Task.all_tasks(loop=loop))


if __name__ == '__main__':
    main()