Removal of unused imports. Added new signal and notification options

Author: Nathan Lee

pyrunner/core/config.py

@@ -103,6 +103,9 @@ def __init__(self):
       'dryrun'             : { 'type': bool, 'preserve': False, 'env': 'APP_DRYRUN'            , 'value': None, 'default': False },
       'email_on_fail'      : { 'type': bool, 'preserve': False, 'env': 'APP_EMAIL_ON_FAIL'     , 'value': None, 'default': True },
       'email_on_success'   : { 'type': bool, 'preserve': False, 'env': 'APP_EMAIL_ON_SUCCESS'  , 'value': None, 'default': True },
+      'notify_on_fail'     : { 'type': bool, 'preserve': False, 'env': 'APP_NOTIFY_ON_FAIL'    , 'value': None, 'default': True },
+      'notify_on_success'  : { 'type': bool, 'preserve': False, 'env': 'APP_NOTIFY_ON_SUCCESS' , 'value': None, 'default': True },
+      'as_service'         : { 'type': bool, 'preserve': True,  'env': 'APP_AS_SERVICE'        , 'value': None, 'default': False },
       'test_mode'          : { 'type': bool, 'preserve': True,  'env': 'APP_TEST_MODE'         , 'value': None, 'default': False }
     }
     self._iter_keys = None

pyrunner/core/engine.py

@@ -17,12 +17,10 @@
 import pyrunner.core.constants as constants
 from pyrunner.core.config import Config
 from pyrunner.core.context import Context
-from pyrunner.core.signal import SignalHandler, SIG_ABORT, SIG_PAUSE
+from pyrunner.core.signal import SignalHandler, SIG_ABORT, SIG_PULSE, SIG_RESTART
 from multiprocessing import Manager
 
-import os, sys, glob
-import time
-import pickle
+import sys, time
 
 class ExecutionEngine:
   """
@@ -89,14 +87,19 @@ def initiate(self, **kwargs):
     # Execution loop
     try:
       while self.register.running_nodes or self.register.pending_nodes:
-        sig_set = signal_handler.consume()
+        # Consume pulse signal, if any, to indicate app is already running
+        signal_handler.consume(SIG_PULSE)
 
         # Check for abort signals
-        if SIG_ABORT in sig_set:
+        if signal_handler.consume(SIG_ABORT):
           print('ABORT signal received! Terminating all running Workers.')
           self._abort_all_workers()
           return -1
 
+        # Check for restart signals; restart failed nodes, if any
+        if signal_handler.consume(SIG_RESTART):
+          pass
+        
         # Poll running nodes for completion/failure
         for node in self.register.running_nodes.copy():
           retcode = node.poll()

pyrunner/core/node.py

@@ -14,18 +14,10 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-import pyrunner.core.constants as constants
 import pyrunner.logger.file as lg
 from pyrunner.worker.abstract import Worker
 
-import os, sys
-import time
-import multiprocessing
-import importlib
-import traceback
-import inspect
-
-from time import gmtime, strftime
+import time, multiprocessing, importlib
 
 class ExecutionNode:
   """

pyrunner/core/pyrunner.py

@@ -19,7 +19,6 @@
 import shutil
 import zipfile
 import getopt
-import traceback
 
 import pyrunner.serde as serde
 import pyrunner.notification as notification
@@ -60,7 +59,7 @@ def __init__(self, **kwargs):
       raise OSError('Another process for "{}" is already running!'.format(self.config['app_name']))
     else:
       # Clear signals, if any, to ensure clean start.
-      self.signal_handler.consume()
+      self.signal_handler.consume_all()
 
   def reset_env(self):
     os.environ.clear()
@@ -360,7 +359,8 @@ def parse_args(self, run_getopts=True):
       'env=', 'cvar=', 'context=', 'time-between-tasks=',
       'to=', 'from=', 'descendants=', 'ancestors=',
       'norun=', 'exec-only=', 'exec-proc-name=',
-      'max-procs=', 'serde=', 'exec-loop-interval='
+      'max-procs=', 'serde=', 'exec-loop-interval=',
+      'notify-on-fail=', 'notify-on-success=', 'as-service'
     ]
 
     if run_getopts:
@@ -396,6 +396,10 @@ def parse_args(self, run_getopts=True):
           self.config['email_on_fail'] = arg
         elif opt in ['--email-on-success']:
           self.config['email_on_success'] = arg
+        elif opt == '--notify-on-fail':
+          self.config['notify_on_fail'] = arg
+        elif opt == '--notify-on-success':
+          self.config['nitory_on_success'] = arg
         elif opt == '--env':
           parts = arg.split('=')
           os.environ[parts[0]] = parts[1]
@@ -420,6 +424,8 @@ def parse_args(self, run_getopts=True):
           self.config['allow_duplicate_jobs'] = True
         elif opt in ['--exec-proc-name']:
           self.config['exec_proc_name'] = arg
+        elif opt == '--as-service':
+          self.config['as_service'] = True
         elif opt == '--abort':
           abort = True
         elif opt == '--silent':

pyrunner/core/signal.py

@@ -19,8 +19,9 @@
 SIG_ABORT = 'sig.abort'
 SIG_PAUSE = 'sig.pause'
 SIG_PULSE = 'sig.pulse'
+SIG_RESTART = 'sig.restart'
 
-_valid_signals = (SIG_ABORT, SIG_PAUSE, SIG_PULSE)
+_valid_signals = (SIG_ABORT, SIG_PAUSE, SIG_PULSE, SIG_RESTART)
 
 class SignalHandler:
 
@@ -31,10 +32,19 @@ def sig_file(self, sig):
     return '{}/.{}.{}'.format(self.config['temp_dir'], self.config['app_name'], sig)
 
   def emit(self, sig):
-    if sig not in _valid_signals: return ValueError('Unknown signal type: {}'.format(sig))
+    if sig not in _valid_signals: raise ValueError('Unknown signal type: {}'.format(sig))
     open(self.sig_file(sig), 'a').close()
 
-  def consume(self):
+  def consume(self, sig):
+    if sig not in _valid_signals:
+      raise ValueError('Unknown signal type: {}'.format(sig))
+    if sig in self.peek():
+      os.remove(self.sig_file(sig))
+      return True
+    else:
+      return False
+  
+  def consume_all(self):
     sig_set = self.peek()
     for sig in sig_set:
       os.remove(self.sig_file(sig))

pyrunner/version.py

@@ -1 +1 @@
-__version__ = '5.2.1'
+__version__ = '6.0.0'