Allow for other PythonScriptEngine environments besides task data

SpiffWorkflow/bpmn/FeelLikeScriptEngine.py

@@ -266,8 +266,8 @@ class FeelLikeScriptEngine(PythonScriptEngine):
     provide a specialised subclass that parses and executes the scripts /
     expressions in a mini-language of your own.
     """
-    def __init__(self):
-        super().__init__()
+    def __init__(self, environment=None):
+        super().__init__(environment=environment)
 
     def validate(self, expression):
         super().validate(self.patch_expression(expression))

SpiffWorkflow/bpmn/PythonScriptEngine.py

@@ -3,7 +3,9 @@
 import copy
 import sys
 import traceback
+import warnings
 
+from .PythonScriptEngineEnvironment import TaskDataEnvironment
 from ..exceptions import SpiffWorkflowException, WorkflowTaskException
 from ..operators import Operator
 
@@ -26,66 +28,6 @@
 # 02110-1301  USA
 
 
-class Box(dict):
-    """
-    Example:
-    m = Box({'first_name': 'Eduardo'}, last_name='Pool', age=24, sports=['Soccer'])
-    """
-
-    def __init__(self, *args, **kwargs):
-        super(Box, self).__init__(*args, **kwargs)
-        for arg in args:
-            if isinstance(arg, dict):
-                for k, v in arg.items():
-                    if isinstance(v, dict):
-                        self[k] = Box(v)
-                    else:
-                        self[k] = v
-
-        if kwargs:
-            for k, v in kwargs.items():
-                if isinstance(v, dict):
-                    self[k] = Box(v)
-                else:
-                    self[k] = v
-
-    def __deepcopy__(self, memodict=None):
-        if memodict is None:
-            memodict = {}
-        my_copy = Box()
-        for k, v in self.items():
-            my_copy[k] = copy.deepcopy(v)
-        return my_copy
-
-    def __getattr__(self, attr):
-        try:
-            output = self[attr]
-        except:
-            raise AttributeError(
-                "Dictionary has no attribute '%s' " % str(attr))
-        return output
-
-    def __setattr__(self, key, value):
-        self.__setitem__(key, value)
-
-    def __setitem__(self, key, value):
-        super(Box, self).__setitem__(key, value)
-        self.__dict__.update({key: value})
-
-    def __getstate__(self):
-        return self.__dict__
-
-    def __setstate__(self, state):
-        self.__init__(state)
-
-    def __delattr__(self, item):
-        self.__delitem__(item)
-
-    def __delitem__(self, key):
-        super(Box, self).__delitem__(key)
-        del self.__dict__[key]
-
-
 class PythonScriptEngine(object):
     """
     This should serve as a base for all scripting & expression evaluation
@@ -97,10 +39,18 @@ class PythonScriptEngine(object):
     expressions in a different way.
     """
 
-    def __init__(self, default_globals=None, scripting_additions=None):
-
-        self.globals = default_globals or {}
-        self.globals.update(scripting_additions or {})
+    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):
@@ -175,53 +125,16 @@ def check_for_overwrite(self, task, external_methods):
         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.globals).intersection(task.data)
+        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 convert_to_box(self, data):
-        if isinstance(data, dict):
-            for key, value in data.items():
-                if not isinstance(value, Box):
-                    data[key] = self.convert_to_box(value)
-            return Box(data)
-        if isinstance(data, list):
-            for idx, value in enumerate(data):
-                data[idx] = self.convert_to_box(value)
-            return data
-        return data
-
     def _evaluate(self, expression, context, external_methods=None):
-
-        globals = copy.copy(self.globals)  # else we pollute all later evals.
-        self.convert_to_box(context)
-        globals.update(external_methods or {})
-        globals.update(context)
-        return eval(expression, globals)
+        return self.environment.evaluate(expression, context, external_methods)
 
     def _execute(self, script, context, external_methods=None):
-
-        my_globals = copy.copy(self.globals)
-        self.convert_to_box(context)
-        my_globals.update(external_methods or {})
-        context.update(my_globals)
-        try:
-            exec(script, context)
-        finally:
-            self.remove_globals_and_functions_from_context(context,
-                                                           external_methods)
-
-    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):
-            if k == "__builtins__" or \
-                    hasattr(context[k], '__call__') or \
-                    k in self.globals or \
-                    external_methods and k in external_methods:
-                context.pop(k)
+        self.environment.execute(script, context, external_methods)

SpiffWorkflow/bpmn/PythonScriptEngineEnvironment.py

@@ -0,0 +1,122 @@
+import copy
+import warnings
+
+class BasePythonScriptEngineEnvironment:
+    def __init__(self, environment_globals=None):
+        self.globals = environment_globals or {}
+
+    def evaluate(self, expression, context, external_methods=None):
+        raise NotImplementedError("Subclass must implement this method")
+
+    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)
+        my_globals.update(external_methods or {})
+        my_globals.update(context)
+        return eval(expression, my_globals)
+
+    def execute(self, script, context, external_methods=None):
+        my_globals = copy.copy(self.globals)
+        self._prepare_context(context)
+        my_globals.update(external_methods or {})
+        context.update(my_globals)
+        try:
+            exec(script, context)
+        finally:
+            self._remove_globals_and_functions_from_context(context, external_methods)
+
+    def _prepare_context(self, context):
+        pass
+
+    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):
+            if k == "__builtins__" or \
+                    hasattr(context[k], '__call__') or \
+                    k in self.globals or \
+                    external_methods and k in external_methods:
+                context.pop(k)
+
+class Box(dict):
+    """
+    Example:
+    m = Box({'first_name': 'Eduardo'}, last_name='Pool', age=24, sports=['Soccer'])
+    """
+
+    def __init__(self, *args, **kwargs):
+        warnings.warn('The usage of Box has been deprecated.', DeprecationWarning, stacklevel=2)
+        super(Box, self).__init__(*args, **kwargs)
+        for arg in args:
+            if isinstance(arg, dict):
+                for k, v in arg.items():
+                    if isinstance(v, dict):
+                        self[k] = Box(v)
+                    else:
+                        self[k] = v
+
+        if kwargs:
+            for k, v in kwargs.items():
+                if isinstance(v, dict):
+                    self[k] = Box(v)
+                else:
+                    self[k] = v
+
+    def __deepcopy__(self, memodict=None):
+        if memodict is None:
+            memodict = {}
+        my_copy = Box()
+        for k, v in self.items():
+            my_copy[k] = copy.deepcopy(v)
+        return my_copy
+
+    def __getattr__(self, attr):
+        try:
+            output = self[attr]
+        except:
+            raise AttributeError(
+                "Dictionary has no attribute '%s' " % str(attr))
+        return output
+
+    def __setattr__(self, key, value):
+        self.__setitem__(key, value)
+
+    def __setitem__(self, key, value):
+        super(Box, self).__setitem__(key, value)
+        self.__dict__.update({key: value})
+
+    def __getstate__(self):
+        return self.__dict__
+
+    def __setstate__(self, state):
+        self.__init__(state)
+
+    def __delattr__(self, item):
+        self.__delitem__(item)
+
+    def __delitem__(self, key):
+        super(Box, self).__delitem__(key)
+        del self.__dict__[key]
+
+    @classmethod
+    def convert_to_box(cls, data):
+        if isinstance(data, dict):
+            for key, value in data.items():
+                if not isinstance(value, Box):
+                    data[key] = cls.convert_to_box(value)
+            return Box(data)
+        if isinstance(data, list):
+            for idx, value in enumerate(data):
+                data[idx] = cls.convert_to_box(value)
+            return data
+        return data
+
+class BoxedTaskDataEnvironment(TaskDataEnvironment):
+    def _prepare_context(self, context):
+        Box.convert_to_box(context)
+

tests/SpiffWorkflow/bpmn/BoxDeepCopyTest.py

@@ -1,6 +1,6 @@
 import unittest
 
-from SpiffWorkflow.bpmn.PythonScriptEngine import Box
+from SpiffWorkflow.bpmn.PythonScriptEngineEnvironment import Box
 
 
 class BoxDeepCopyTest(unittest.TestCase):

tests/SpiffWorkflow/bpmn/CustomScriptTest.py

@@ -4,6 +4,7 @@
 from SpiffWorkflow.exceptions import WorkflowTaskException
 from SpiffWorkflow.task import TaskState
 from SpiffWorkflow.bpmn.PythonScriptEngine import PythonScriptEngine
+from SpiffWorkflow.bpmn.PythonScriptEngineEnvironment import TaskDataEnvironment
 from SpiffWorkflow.bpmn.workflow import BpmnWorkflow
 from tests.SpiffWorkflow.bpmn.BpmnWorkflowTestCase import BpmnWorkflowTestCase
 
@@ -17,8 +18,8 @@ class CustomBpmnScriptEngine(PythonScriptEngine):
     It will execute python code read in from the bpmn.  It will also make any scripts in the
      scripts directory available for execution. """
     def __init__(self):
-        augment_methods = {'custom_function': my_custom_function}
-        super().__init__(scripting_additions=augment_methods)
+        environment = TaskDataEnvironment({'custom_function': my_custom_function})
+        super().__init__(environment=environment)
 
 
 class CustomInlineScriptTest(BpmnWorkflowTestCase):

tests/SpiffWorkflow/bpmn/FeelExpressionEngineTest.py

@@ -3,6 +3,7 @@
 import unittest
 
 from SpiffWorkflow.bpmn.FeelLikeScriptEngine import FeelLikeScriptEngine, FeelInterval
+from SpiffWorkflow.bpmn.PythonScriptEngineEnvironment import BoxedTaskDataEnvironment
 from tests.SpiffWorkflow.bpmn.BpmnWorkflowTestCase import BpmnWorkflowTestCase
 import datetime
 
@@ -12,7 +13,7 @@
 class FeelExpressionTest(BpmnWorkflowTestCase):
 
     def setUp(self):
-        self.expressionEngine = FeelLikeScriptEngine()
+        self.expressionEngine = FeelLikeScriptEngine(environment=BoxedTaskDataEnvironment())
 
     def testRunThroughExpressions(self):
         tests = [("string length('abcd')", 4, {}),
@@ -62,7 +63,7 @@ def testRunThroughDMNExpression(self):
             ]
         }
         x = self.expressionEngine._evaluate(
-            """sum([1 for x in exclusive if x.get('ExclusiveSpaceAMComputingID',None)==None])""", 
+            """sum([1 for x in exclusive if x.get('ExclusiveSpaceAMComputingID',None)==None])""",
             data
         )
         self.assertEqual(x, 1)