When schedule a job,
- you want to make sure the scheduled job runs successfully.
- And you need to make sure that schedules won't affect other schedules if all of them run in parallel.
- You need some try catch statement to keep it running, in the mean while, keep track of the error.
So you can use the following functions to do that. By doing so, the error of one job will be written in a log.txt file, in the same time, other jobs will keep running.
import functools
import os
import time
import logging
import schedule
from multiprocessing import *
from datetime import datetime
from subprocess import call, PIPE, Popen, TimeoutExpired
# initialize the log settings
logging.basicConfig(filename='schedule_permit.log', level=logging.INFO)
def pre_run_job(py_file):
stderr = '' cmd = ['python', py_file]
p = Popen(cmd, shell=True, stdout=PIPE, stderr=PIPE)
try:
# let the job run 30 seconds to make sure it works fine
stdout, stderr = p.communicate(timeout=30)
except TimeoutExpired:
p.kill() # stop the job after 30 seconds
return stderr
def with_logging(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
print('LOG: Running job "%s"' % func.__name__)
print('Started at ' + str(datetime.now()))
result = func(*args, **kwargs)
print('LOG: Job "%s" Stopped' % func.__name__)
print('Stopped at ' + str(datetime.now()))
return result
return wrapper
@with_logging
def run_test():
try:
# Return error. if no error, then run the schedule
err_msg = pre_run_job(get_file('tx/permit/permit.py'))
print(err_msg)
logging.exception(err_msg)
if err_msg == '':
print('file is OK. Now start the schedule.')
schedule.every(1).seconds.do(call, ['python', get_file('tx/permit/permit.py')])
run_schedule()
except Exception as e:
logging.exception(e)
@with_loggingdef run_test2():
try:
schedule.every(10).seconds.do(print, 'hello')
run_schedule()
except Exception as e:
logging.exception(e)
@with_loggingdef run_job():
List_of_processes = [Process(target=run_test), Process(target=run_test2)]
for p in List_of_processes:
p.start()
for p in List_of_processes:
p.join()
if __name__ == "__main__":
run_job()
No comments:
Post a Comment