Removal of legacy style workers. Removed generate_worker and SHELL mode process list option. Patched major file descriptor leak due to joined proc handles not being GC'ed.

Author: Nathan Lee

README.md

@@ -1,4 +1,6 @@
 # python-batch-runner
+[![Documentation Status](https://readthedocs.org/projects/python-batch-runner/badge/?version=latest)](https://python-batch-runner.readthedocs.io/en/latest/?badge=latest)
+
 For more complete documentation, please see: https://python-batch-runner.readthedocs.io/
 
 python-batch-runner is a microframework to assist with building small to medium scale batch applications without needing to build the scaffolding from scratch.

pyrunner/cli.py

@@ -127,7 +127,7 @@ def setup():
     app_profile.write('export APP_EXEC_TIMESTAMP=$(date +"%Y%m%d_%H%M%S")\n\n')
     app_profile.write('export APP_LOG_DIR="${APP_ROOT_LOG_DIR}/${DATE}"\n\n')
     app_profile.write('if [ ! -e ${APP_LOG_DIR}  ]; then mkdir -p ${APP_LOG_DIR}; fi\n')
-    app_profile.write('if [ ! -e ${APP_TEMP_DIR} ]; then mkdir ${APP_TEMP_DIR}; fi\n')
+    app_profile.write('if [ ! -e ${APP_TEMP_DIR} ]; then mkdir -p ${APP_TEMP_DIR}; fi\n')
 
   print('Creating Blank Process List File: {}/config/{}.lst'.format(app_root, app_name))
   with open('{}/config/{}.lst'.format(app_root, app_name), 'w') as lst_file:

pyrunner/core/constants.py

@@ -28,10 +28,7 @@
 
 EXECUTION_TIMESTAMP = datetime.now().strftime("%Y%m%d_%H%M%S")
 
-MODE_SHELL  = 'SHELL'
 MODE_PYTHON = 'PYTHON'
-
-HEADER_SHELL  = '#{}\n#ID|PARENT_IDS|MAX_ATTEMPTS|RETRY_WAIT_TIME|PROCESS_NAME|SHELL_COMMAND|LOGFILE'.format(MODE_SHELL)
 HEADER_PYTHON = '#{}\n#ID|PARENT_IDS|MAX_ATTEMPTS|RETRY_WAIT_TIME|PROCESS_NAME|MODULE_NAME|WORKER_NAME|ARGUMENTS|LOGFILE'.format(MODE_PYTHON)
 
 ROOT_NODE_NAME = 'PyRunnerRootNode'
@@ -40,7 +37,6 @@
 
 import os, sys
 from pyrunner import PyRunner
-from pathlib import Path
 
 # Determine absolute path of this file's parent directory at runtime
 abs_dir_path = os.path.dirname(os.path.realpath(__file__))

pyrunner/core/node.py

@@ -110,7 +110,7 @@ def execute(self):
       # If it does not extend the Worker class, initialize a reverse-Worker in which the
       # worker extends the provided class.
       else:
-        worker = self.generate_worker()(self.context, self._retcode, self.logfile, self.argv)
+        raise TypeError('{}.{} is not an extension of pyrunner.Worker'.format(self.module, self.worker))
 
       # Launch the "run" method of the provided Worker under a new process.
       self._thread = multiprocessing.Process(target=worker.protected_run, daemon=False)
@@ -146,6 +146,7 @@ def poll(self, wait=False):
       # causing the thread to block until it's job is complete.
       self._thread.join()
       self._end_time = time.time()
+      self._thread = None
       if self.retcode > 0:
         if self._attempts < self.max_attempts:
           logger = lg.FileLogger(self.logfile)
@@ -154,6 +155,7 @@ def poll(self, wait=False):
           self._must_wait = True
           self._wait_start = time.time()
           logger.restart_message(self._attempts)
+          logger.close()
           self._retcode.value = -1
     elif (time.time() - self._start_time) >= self._timeout:
       self._thread.terminate()
@@ -233,100 +235,6 @@ def get_elapsed_time(self):
     else:
       return '00:00:00'
 
-  # ########################## GENERATE WORKER ########################## #
-  
-  def generate_worker(self):
-    """
-    * For backwards compatibility with earlier versions. *
-    
-    Returns a generic Worker object which extends the user-defined parent
-    class. This is done in order to expose the context, logger, and argv
-    attributes to the user-defined worker.
-    """
-    parent_class = getattr(importlib.import_module(self.module), self.worker)
-    
-    class Worker(parent_class):
-      
-      def __init__(self, context, retcode, logfile, argv):
-        self._context = context
-        self._retcode = retcode
-        self.logfile = logfile
-        self.logger = lg.FileLogger(logfile).open()
-        self.argv = argv
-        return
-      
-      @property
-      def context(self):
-        return getattr(self, '_context', None)
-      @context.setter
-      def context(self, value):
-        self._context = value
-        return self
-      
-      @property
-      def retcode(self):
-        return self._retcode.value
-      @retcode.setter
-      def retcode(self, value):
-        if int(value) < 0:
-          raise ValueError('retcode must be 0 or greater - received: {}'.format(value))
-        self._retcode.value = int(value)
-        return self
-      
-      # TODO: Need to deprecate
-      def set_return_code(self, value):
-        self.retcode = int(value)
-        return
-      
-      def protected_run(self):
-        '''Initiate worker class run method and additionally trigger methods if defined
-        for other lifecycle steps.'''
-        
-        # RUN
-        try:
-          self.retcode = super().run() or self.retcode
-        except Exception as e:
-          self.logger.error("Uncaught Exception from Worker Thread (RUN)")
-          self.logger.error(str(e))
-          self.logger.error(traceback.format_exc())
-          self.retcode = 901
-        
-        if not self.retcode:
-          # ON SUCCESS
-          if parent_class.__dict__.get('on_success'):
-            try:
-              self.retcode = super().on_success() or self.retcode
-            except Exception as e:
-              self.logger.error('Uncaught Exception from Worker Thread (ON_SUCCESS)')
-              self.logger.error(str(e))
-              self.logger.error(traceback.format_exc())
-              self.retcode = 902
-        else:
-          # ON FAIL
-          if parent_class.__dict__.get('on_fail'):
-            try:
-              self.retcode = super().on_fail() or self.retcode
-            except Exception as e:
-              self.logger.error('Uncaught Exception from Worker Thread (ON_FAIL)')
-              self.logger.error(str(e))
-              self.logger.error(traceback.format_exc())
-              self.retcode = 903
-        
-        # ON EXIT
-        if parent_class.__dict__.get('on_exit'):
-          try:
-            self.retcode = super().on_exit() or self.retcode
-          except Exception as e:
-            self.logger.error('Uncaught Exception from Worker Thread (ON_EXIT)')
-            self.logger.error(str(e))
-            self.logger.error(traceback.format_exc())
-            self.retcode = 904
-        
-        self.logger.close()
-        
-        return
-    
-    return Worker
 
   # ########################## SETTERS + GETTERS ########################## #
 

pyrunner/logger/file.py

@@ -69,6 +69,7 @@ def close(self):
       self.logfile_handle.write("# LOG END - {}\n".format(datetime.now()))
       self.logfile_handle.write("############################################################################\n\n")
       self.logfile_handle.close()
+      self.logfile_handle = None
     return
 
   def get_file_handle(self):

pyrunner/serde/list.py

@@ -35,21 +35,6 @@ def deserialize(self, proc_file, restart=False):
 
     if not proc_list: raise ValueError('No information read from process list file')
 
-    i = 0
-    header = ''
-    
-    while not header:
-      header = proc_list[i].strip()
-      i += 1
-    
-    if header[0] != '#':
-      raise RuntimeError('Missing execution mode header in process list file. Must have at minimum:\n#SHELL\nor\n#PYTHON')
-      
-    mode = header[1:].split('|')[0]
-    
-    if mode not in [ constants.MODE_SHELL, constants.MODE_PYTHON ]:
-      raise RuntimeError('Incorrect execution mode in header: {}'.format(mode))
-    
     used_ids = set()
 
     for proc in proc_list:
@@ -85,62 +70,33 @@ def deserialize(self, proc_file, restart=False):
 
       dependencies = [ int(x) for x in sub_details[1].split(',') ]
 
-      if mode == constants.MODE_SHELL:
-        if restart:
-          register.add_node(
-            id = id,
-            dependencies = dependencies,
-            max_attempts = sub_details[2],
-            retry_wait_time = sub_details[3],
-            status = sub_details[4] if sub_details[4] in [ constants.STATUS_COMPLETED, constants.STATUS_NORUN ] else constants.STATUS_PENDING,
-            name = sub_details[6],
-            module = 'pyrunner',
-            worker = 'ShellWorker',
-            arguments = [sub_details[7]],
-            logfile = sub_details[8] if len(sub_details) > 8 else None,
-            named_deps = False
-          )
-        else:
-          register.add_node(
-            id = id,
-            dependencies = dependencies,
-            max_attempts = sub_details[2],
-            retry_wait_time = sub_details[3],
-            name = sub_details[4],
-            module = 'pyrunner',
-            worker = 'ShellWorker',
-            arguments = [sub_details[5]],
-            logfile = sub_details[6] if len(sub_details) > 6 else None,
-            named_deps = False
-          )
+      if restart:
+        register.add_node(
+          id = id,
+          dependencies = dependencies,
+          max_attempts = sub_details[2],
+          retry_wait_time = sub_details[3],
+          status = sub_details[4] if sub_details[4] in [ constants.STATUS_COMPLETED, constants.STATUS_NORUN ] else constants.STATUS_PENDING,
+          name = sub_details[6],
+          module = sub_details[7],
+          worker = sub_details[8],
+          arguments = [ s.strip('"') if s.strip().startswith('"') and s.strip().endswith('"') else s.strip() for s in comma_pattern.split(sub_details[9])[1::2] ] if len(sub_details) > 9 else None,
+          logfile = sub_details[10] if len(sub_details) > 10 else None,
+          named_deps = False
+        )
       else:
-        if restart:
-          register.add_node(
-            id = id,
-            dependencies = dependencies,
-            max_attempts = sub_details[2],
-            retry_wait_time = sub_details[3],
-            status = sub_details[4] if sub_details[4] in [ constants.STATUS_COMPLETED, constants.STATUS_NORUN ] else constants.STATUS_PENDING,
-            name = sub_details[6],
-            module = sub_details[7],
-            worker = sub_details[8],
-            arguments = [ s.strip('"') if s.strip().startswith('"') and s.strip().endswith('"') else s.strip() for s in comma_pattern.split(sub_details[9])[1::2] ] if len(sub_details) > 9 else None,
-            logfile = sub_details[10] if len(sub_details) > 10 else None,
-            named_deps = False
-          )
-        else:
-          register.add_node(
-            id = id,
-            dependencies = dependencies,
-            max_attempts = sub_details[2],
-            retry_wait_time = sub_details[3],
-            name = sub_details[4],
-            module = sub_details[5],
-            worker = sub_details[6],
-            arguments = [ s.strip('"') if s.strip().startswith('"') and s.strip().endswith('"') else s.strip() for s in comma_pattern.split(sub_details[7])[1::2] ] if len(sub_details) > 7 else None,
-            logfile = sub_details[8] if len(sub_details) > 8 else None,
-            named_deps = False
-          )
+        register.add_node(
+          id = id,
+          dependencies = dependencies,
+          max_attempts = sub_details[2],
+          retry_wait_time = sub_details[3],
+          name = sub_details[4],
+          module = sub_details[5],
+          worker = sub_details[6],
+          arguments = [ s.strip('"') if s.strip().startswith('"') and s.strip().endswith('"') else s.strip() for s in comma_pattern.split(sub_details[7])[1::2] ] if len(sub_details) > 7 else None,
+          logfile = sub_details[8] if len(sub_details) > 8 else None,
+          named_deps = False
+        )
 
     return register
 

pyrunner/version.py

@@ -1 +1 @@
-__version__ = '4.5.0'
+__version__ = '5.0.0'