Merge pull request #315 from sartography/feature/new-task-states

Feature/new task states

Author: essweine

SpiffWorkflow/bpmn/PythonScriptEngine.py

@@ -1,13 +1,11 @@
 # -*- coding: utf-8 -*-
 import ast
-import copy
 import sys
 import traceback
 import warnings
 
 from .PythonScriptEngineEnvironment import TaskDataEnvironment
 from ..exceptions import SpiffWorkflowException, WorkflowTaskException
-from ..operators import Operator
 
 
 # Copyright (C) 2020 Kelly McDonald
@@ -40,18 +38,19 @@ class PythonScriptEngine(object):
     """
 
     def __init__(self, default_globals=None, scripting_additions=None, environment=None):
+
         if default_globals is not None or scripting_additions is not None:
             warnings.warn(f'default_globals and scripting_additions are deprecated.  '
                           f'Please provide an environment such as TaskDataEnvrionment',
                           DeprecationWarning, stacklevel=2)
+
         if environment is None:
             environment_globals = {}
             environment_globals.update(default_globals or {})
             environment_globals.update(scripting_additions or {})
             self.environment = TaskDataEnvironment(environment_globals)
         else:
             self.environment = environment
-        self.error_tasks = {}
 
     def validate(self, expression):
         ast.parse(expression)
@@ -62,28 +61,19 @@ def evaluate(self, task, expression, external_methods=None):
         return the result.
         """
         try:
-            if isinstance(expression, Operator):
-                # I am assuming that this takes care of some kind of XML
-                # expression judging from the contents of operators.py
-                return expression._matches(task)
-            else:
-                return self._evaluate(expression, task.data, external_methods)
+            return self._evaluate(expression, task.data, external_methods)
         except SpiffWorkflowException as se:
             se.add_note(f"Error evaluating expression '{expression}'")
             raise se
         except Exception as e:
             raise WorkflowTaskException(f"Error evaluating expression '{expression}'", task=task, exception=e)
 
     def execute(self, task, script, external_methods=None):
-        """
-        Execute the script, within the context of the specified task
-        """
+        """Execute the script, within the context of the specified task."""
         try:
-            self.check_for_overwrite(task, external_methods or {})
-            self._execute(script, task.data, external_methods or {})
+            return self._execute(script, task.data, external_methods or {})
         except Exception as err:
             wte = self.create_task_exec_exception(task, script, err)
-            self.error_tasks[task.id] = wte
             raise wte
 
     def call_service(self, operation_name, operation_params, task_data):
@@ -120,21 +110,8 @@ def get_error_line_number_and_content(self, script, err):
             error_line = script.splitlines()[line_number - 1]
         return line_number, error_line
 
-    def check_for_overwrite(self, task, external_methods):
-        """It's possible that someone will define a variable with the
-        same name as a pre-defined script, rending the script un-callable.
-        This results in a nearly indecipherable error.  Better to fail
-        fast with a sensible error message."""
-        func_overwrites = set(self.environment.globals).intersection(task.data)
-        func_overwrites.update(set(external_methods).intersection(task.data))
-        if len(func_overwrites) > 0:
-            msg = f"You have task data that overwrites a predefined " \
-                  f"function(s). Please change the following variable or " \
-                  f"field name(s) to something else: {func_overwrites}"
-            raise WorkflowTaskException(msg, task=task)
-
     def _evaluate(self, expression, context, external_methods=None):
         return self.environment.evaluate(expression, context, external_methods)
 
     def _execute(self, script, context, external_methods=None):
-        self.environment.execute(script, context, external_methods)
+        return self.environment.execute(script, context, external_methods)

SpiffWorkflow/bpmn/PythonScriptEngineEnvironment.py

@@ -1,6 +1,7 @@
 import copy
 import warnings
 
+
 class BasePythonScriptEngineEnvironment:
     def __init__(self, environment_globals=None):
         self.globals = environment_globals or {}
@@ -11,7 +12,9 @@ def evaluate(self, expression, context, external_methods=None):
     def execute(self, script, context, external_methods=None):
         raise NotImplementedError("Subclass must implement this method")
 
+
 class TaskDataEnvironment(BasePythonScriptEngineEnvironment):
+
     def evaluate(self, expression, context, external_methods=None):
         my_globals = copy.copy(self.globals)  # else we pollute all later evals.
         self._prepare_context(context)
@@ -20,6 +23,7 @@ def evaluate(self, expression, context, external_methods=None):
         return eval(expression, my_globals)
 
     def execute(self, script, context, external_methods=None):
+        self.check_for_overwrite(context, external_methods or {})
         my_globals = copy.copy(self.globals)
         self._prepare_context(context)
         my_globals.update(external_methods or {})
@@ -28,12 +32,12 @@ def execute(self, script, context, external_methods=None):
             exec(script, context)
         finally:
             self._remove_globals_and_functions_from_context(context, external_methods)
+        return True
 
     def _prepare_context(self, context):
         pass
 
-    def _remove_globals_and_functions_from_context(self, context,
-                                                  external_methods=None):
+    def _remove_globals_and_functions_from_context(self, context, external_methods=None):
         """When executing a script, don't leave the globals, functions
         and external methods in the context that we have modified."""
         for k in list(context):
@@ -43,6 +47,20 @@ def _remove_globals_and_functions_from_context(self, context,
                     external_methods and k in external_methods:
                 context.pop(k)
 
+    def check_for_overwrite(self, context, external_methods):
+        """It's possible that someone will define a variable with the
+        same name as a pre-defined script, rendering the script un-callable.
+        This results in a nearly indecipherable error.  Better to fail
+        fast with a sensible error message."""
+        func_overwrites = set(self.globals).intersection(context)
+        func_overwrites.update(set(external_methods).intersection(context))
+        if len(func_overwrites) > 0:
+            msg = f"You have task data that overwrites a predefined " \
+                  f"function(s). Please change the following variable or " \
+                  f"field name(s) to something else: {func_overwrites}"
+            raise ValueError(msg)
+
+
 class Box(dict):
     """
     Example:

SpiffWorkflow/bpmn/serializer/migration/version_1_2.py

@@ -16,58 +16,64 @@ def convert_timer_expressions(dct):
 
     message = "Unable to convert time specifications for {spec}. This most likely because the values are set during workflow execution."
 
+    # Moving this code into helper functions to make sonarcloud STFU about this file.
+    # Don't really consider this better but whatever.
+
+    def convert_timedate(spec):
+        expr = spec['event_definition'].pop('dateTime')
+        try:
+            dt = eval(expr)
+            if isinstance(dt, datetime):
+                spec['event_definition']['expression'] = f"'{dt.isoformat()}'"
+                spec['event_definition']['typename'] = 'TimeDateEventDefinition'
+            elif isinstance(dt, timedelta):
+                spec['event_definition']['expression'] = f"'{td_to_iso(dt)}'"
+                spec['event_definition']['typename'] = 'DurationTimerEventDefinition'
+        except:
+            raise VersionMigrationError(message.format(spec=spec['name']))
+
+    def convert_cycle(spec, task):
+        expr = spec['event_definition'].pop('cycle_definition')
+        try:
+            repeat, duration = eval(expr)
+            spec['event_definition']['expression'] = f"'R{repeat}/{td_to_iso(duration)}'"
+            if task is not None:
+                cycles_complete = task['data'].pop('repeat_count', 0)
+                start_time = task['internal_data'].pop('start_time', None)
+                if start_time is not None:
+                    dt = datetime.fromisoformat(start_time)
+                    task['internal_data']['event_value'] = {
+                        'cycles': repeat - cycles_complete,
+                        'next': datetime.combine(dt.date(), dt.time(), LOCALTZ).isoformat(),
+                        'duration': duration.total_seconds(),
+                    }
+        except:
+            raise VersionMigrationError(message.format(spec=spec['name']))
+
+        if spec['typename'] == 'StartEvent':
+            spec['outputs'].remove(spec['name'])
+            if task is not None:
+                children = [ dct['tasks'][c] for c in task['children'] ]
+                # Formerly cycles were handled by looping back and reusing the tasks so this removes the extra tasks
+                remove = [ c for c in children if c['task_spec'] == task['task_spec']][0]
+                for task_id in remove['children']:
+                    child = dct['tasks'][task_id]
+                    if child['task_spec'].startswith('return') or child['state'] != TaskState.COMPLETED:
+                        dct['tasks'].pop(task_id)
+                    else:
+                        task['children'].append(task_id)
+                task['children'].remove(remove['id'])
+                dct['tasks'].pop(remove['id'])
+
     has_timer = lambda ts: 'event_definition' in ts and ts['event_definition']['typename'] in [ 'CycleTimerEventDefinition', 'TimerEventDefinition']
     for spec in [ ts for ts in dct['spec']['task_specs'].values() if has_timer(ts) ]:
         spec['event_definition']['name'] = spec['event_definition'].pop('label')
         if spec['event_definition']['typename'] == 'TimerEventDefinition':
-            expr = spec['event_definition'].pop('dateTime')
-            try:
-                dt = eval(expr)
-                if isinstance(dt, datetime):
-                    spec['event_definition']['expression'] = f"'{dt.isoformat()}'"
-                    spec['event_definition']['typename'] = 'TimeDateEventDefinition'
-                elif isinstance(dt, timedelta):
-                    spec['event_definition']['expression'] = f"'{td_to_iso(dt)}'"
-                    spec['event_definition']['typename'] = 'DurationTimerEventDefinition'
-            except:
-                raise VersionMigrationError(message.format(spec=spec['name']))
-
+            convert_timedate(spec)
         if spec['event_definition']['typename'] == 'CycleTimerEventDefinition':
-
             tasks = [ t for t in dct['tasks'].values() if t['task_spec'] == spec['name'] ]
             task = tasks[0] if len(tasks) > 0 else None
-
-            expr = spec['event_definition'].pop('cycle_definition')
-            try:
-                repeat, duration = eval(expr)
-                spec['event_definition']['expression'] = f"'R{repeat}/{td_to_iso(duration)}'"
-                if task is not None:
-                    cycles_complete = task['data'].pop('repeat_count', 0)
-                    start_time = task['internal_data'].pop('start_time', None)
-                    if start_time is not None:
-                        dt = datetime.fromisoformat(start_time)
-                        task['internal_data']['event_value'] = {
-                            'cycles': repeat - cycles_complete,
-                            'next': datetime.combine(dt.date(), dt.time(), LOCALTZ).isoformat(),
-                            'duration': duration.total_seconds(),
-                        }
-            except:
-                raise VersionMigrationError(message.format(spec=spec['name']))
-
-            if spec['typename'] == 'StartEvent':
-                spec['outputs'].remove(spec['name'])
-                if task is not None:
-                    children = [ dct['tasks'][c] for c in task['children'] ]
-                    # Formerly cycles were handled by looping back and reusing the tasks so this removes the extra tasks
-                    remove = [ c for c in children if c['task_spec'] == task['task_spec']][0]
-                    for task_id in remove['children']:
-                        child = dct['tasks'][task_id]
-                        if child['task_spec'].startswith('return') or child['state'] != TaskState.COMPLETED:
-                            dct['tasks'].pop(task_id)
-                        else:
-                            task['children'].append(task_id)
-                    task['children'].remove(remove['id'])
-                    dct['tasks'].pop(remove['id'])
+            convert_cycle(spec, task)
 
 def add_default_condition_to_cond_task_specs(dct):
 
@@ -122,3 +128,18 @@ def remove_loop_reset(dct):
                 parent = dct['tasks'].get(task['parent'])
                 parent['children'] = [c for c in parent['children'] if c != task['id']]
         dct['spec']['task_specs'].pop(spec['name'])
+
+def update_task_states(dct):
+
+    def update(process):
+        for task in process['tasks'].values():
+            if task['state'] == 32:
+                task['state'] = TaskState.COMPLETED
+            elif task['state'] == 64:
+                task['state'] = TaskState.CANCELLED
+
+    root = dct['tasks'].get(dct['root'])
+    if root['state'] == 32:
+        update(dct)
+        for sp in dct['subprocesses'].values():
+            update(sp)

SpiffWorkflow/bpmn/serializer/migration/version_migration.py

@@ -7,6 +7,7 @@
     create_data_objects_and_io_specs,
     check_multiinstance,
     remove_loop_reset,
+    update_task_states,
 )
 
 def from_version_1_1(old):
@@ -36,6 +37,7 @@ def from_version_1_1(old):
     create_data_objects_and_io_specs(new)
     check_multiinstance(new)
     remove_loop_reset(new)
+    update_task_states(new)
     new['VERSION'] = "1.2"
     return new
 
@@ -53,6 +55,7 @@ def from_version_1_0(old):
     attributes based on the task states.
     """
     new = deepcopy(old)
+    new['VERSION'] = "1.1"
     move_subprocesses_to_top(new)
     return from_version_1_1(new)
 

SpiffWorkflow/bpmn/specs/ScriptTask.py

@@ -18,7 +18,6 @@
 # 02110-1301  USA
 
 from .BpmnSpecMixin import BpmnSpecMixin
-from ...task import TaskState
 from ...specs.Simple import Simple
 
 
@@ -30,13 +29,8 @@ def _execute(self, task):
         pass
 
     def _run_hook(self, task):
-        try:
-            self._execute(task)
-            super(ScriptEngineTask, self)._run_hook(task)
-        except Exception as exc:
-            task._set_state(TaskState.WAITING)
-            raise exc
-        return True
+        return self._execute(task)
+
 
 class ScriptTask(ScriptEngineTask):
 
@@ -54,5 +48,4 @@ def spec_type(self):
         return 'Script Task'
 
     def _execute(self, task):
-        task.workflow.script_engine.execute(task, self.script)
-
+        return task.workflow.script_engine.execute(task, self.script)

SpiffWorkflow/serializer/dict.py

@@ -21,7 +21,7 @@
 from base64 import b64encode, b64decode
 from ..workflow import Workflow
 from ..util.impl import get_class
-from ..task import Task
+from ..task import Task, TaskState
 from ..operators import (Attrib, PathAttrib, Equal, NotEqual, Operator, GreaterThan, LessThan, Match)
 from ..specs.base import TaskSpec
 from ..specs.AcquireMutex import AcquireMutex
@@ -604,11 +604,18 @@ def deserialize_workflow(self, s_state, wf_class=Workflow, wf_spec=None, **kwarg
         workflow.spec = wf_spec
         workflow.task_tree = self.deserialize_task(workflow, s_state['task_tree'], reset_specs)
 
-        # Re-connect parents
+        # Re-connect parents and update states if necessary
         tasklist = workflow.get_tasks()
+        root = workflow.get_tasks_from_spec_name('Root')[0]
+        update_state = root.state != TaskState.COMPLETED
         for task in tasklist:
             if task.parent is not None:
                 task.parent = workflow.get_task_from_id(task.parent, tasklist)
+            if update_state:
+                if task.state == 32:
+                    task.state = TaskState.COMPLETED
+                elif task.state == 64:
+                    task.state = TaskState.CANCELLED
 
         if workflow.last_task is not None:
             workflow.last_task = workflow.get_task_from_id(s_state['last_task'],tasklist)