Thursday, January 30, 2020

Python - Capture error from one python script to another

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