Executor: add create_workflow

The `create_workflow` method must be called at the start of
`Executor.run`.

`run` can later set the `_workflow` instance to None, if this preserves
memory...

Signed-off-by: Yves Bastide <yves@botify.com>

Author: ybastide

simpleflow/executor.py

@@ -51,16 +51,23 @@ def __init__(self, workflow_class):
     def workflow_class(self):
         return self._workflow_class
 
+    @property
+    def workflow(self):
+        return self._workflow
+
+    def create_workflow(self):
+        if self._workflow is None:
+            workflow = self._workflow_class(self)
+            if False:
+                assert isinstance(workflow, Workflow)
+            self._workflow = workflow
+
     def run_workflow(self, *args, **kwargs):
         """
         Runs the workflow definition.
 
         """
-        workflow = self.workflow_class(self)
-        if False:
-            assert isinstance(workflow, Workflow)
-        self._workflow = workflow
-        result = workflow.run(*args, **kwargs)
+        result = self._workflow.run(*args, **kwargs)
         return result
 
     @abc.abstractmethod

simpleflow/local/executor.py

@@ -105,6 +105,7 @@ def run(self, input=None):
             input = {}
         args = input.get('args', ())
         kwargs = input.get('kwargs', {})
+        self.create_workflow()
 
         self.initialize_history(input)
 

simpleflow/swf/executor.py

@@ -188,6 +188,7 @@ def reset(self):
         self._idempotent_tasks_to_submit = set()
         self._execution = None
         self.current_priority = None
+        self.create_workflow()
 
     def _make_task_id(self, a_task, *args, **kwargs):
         """
@@ -784,6 +785,7 @@ def replay(self, decision_response):
                 len(self._decisions),
             ))
             self.after_replay()
+            self.decref_workflow()
             return self._decisions, {}
         except exceptions.TaskException as err:
             reason = 'Workflow execution error in task {}: "{}"'.format(
@@ -800,6 +802,7 @@ def replay(self, decision_response):
                 details=swf.format.details(details),
             )
             self.after_closed()
+            self.decref_workflow()
             return [decision], {}
 
         except Exception as err:
@@ -820,15 +823,20 @@ def replay(self, decision_response):
                 details=swf.format.details(details),
             )
             self.after_closed()
+            self.decref_workflow()
             return [decision], {}
 
         self.after_replay()
         decision = swf.models.decision.WorkflowExecutionDecision()
         decision.complete(result=swf.format.result(json_dumps(result)))
         self.on_completed()
         self.after_closed()
+        self.decref_workflow()
         return [decision], {}
 
+    def decref_workflow(self):
+        self._workflow = None
+
     def before_replay(self):
         return self._workflow.before_replay(self._history)
 

simpleflow/swf/task.py

@@ -32,7 +32,6 @@ def schedule(self, domain, task_list=None, **kwargs):
         :param task_list:
         :type task_list: Optional[str]
         :param kwargs:
-        :type kwargs: dict
         :return:
         :rtype: list[swf.models.decision.Decision]
         """

tests/test_simpleflow/test_dataflow.py

@@ -18,7 +18,6 @@
 from swf.responses import Response
 
 from simpleflow import (
-    Workflow,
     futures,
 )
 from simpleflow.task import ActivityTask
@@ -40,6 +39,9 @@
 )
 
 
+Executor.decref_workflow = lambda _: None  # FIXME use @patch?
+
+
 def check_task_scheduled_decision(decision, task):
     """
     Asserts that *decision* schedules *task*.