WIP, does not compute

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

Author: ybastide

simpleflow/executor.py

@@ -3,9 +3,8 @@
 
 from ._decorators import deprecated
 
-
 if False:
-    from typing import Text, Type
+    from typing import Type
     from simpleflow import Workflow
 
 __all__ = ['Executor']
@@ -32,9 +31,6 @@ class Executor(object):
     - asynchronous
     - asynchronous with full replay
 
-     :ivar _workflow: the workflow
-     :type _workflow: simpleflow.workflow.Workflow
-
     """
     __metaclass__ = abc.ABCMeta
 
@@ -48,22 +44,22 @@ def __init__(self, workflow_class):
         as a program, the workflow, and an interpreter, the executor.
 
         """
-        self._workflow = workflow_class(self)
+        self._workflow_class = workflow_class
+        self._workflow = None
 
     @property
-    def workflow(self):
-        """
-        :return:
-        :rtype: simpleflow.workflow.Workflow
-        """
-        return self._workflow
+    def workflow_class(self):
+        return self._workflow_class
 
     def run_workflow(self, *args, **kwargs):
         """
         Runs the workflow definition.
 
         """
-        workflow = self._workflow
+        workflow = self.workflow_class(self)
+        if False:
+            assert isinstance(workflow, Workflow)
+        self._workflow = workflow
         result = workflow.run(*args, **kwargs)
         return result
 
@@ -84,7 +80,7 @@ def submit(self, submittable, *args, **kwargs):
     def map(self, callable, iterable):
         """Submit *callable* with each of the items in ``*iterables``.
 
-        All items in ``*iterables`` must be serializable in JSON.
+        All items in ``*iterable`` must be serializable in JSON.
 
         """
         return [self.submit(callable, argument) for

simpleflow/local/executor.py

@@ -24,18 +24,18 @@ class Executor(executor.Executor):
     """
     def __init__(self, workflow_class):
         super(Executor, self).__init__(workflow_class)
+        self.update_workflow_class()
         self.nb_activities = 0
         self.signals_sent = set()
 
-    @property
-    def _workflow_class(self):
+    def update_workflow_class(self):
         """
         Returns the workflow class with all the needed attributes for
         swf.models.history.builder.History()
         This allows to get a SWF-compatible history in local executions so that
         the metrology feature works correctly.
         """
-        cls = self._workflow.__class__
+        cls = self._workflow_class
         for attr in ("decision_tasks_timeout", "execution_timeout", ):
             if not hasattr(cls, attr):
                 setattr(cls, attr, None)

simpleflow/swf/executor.py

@@ -21,7 +21,6 @@
 from simpleflow.base import Submittable
 from simpleflow.history import History
 from simpleflow.signal import WaitForSignal
-from simpleflow.swf import constants
 from simpleflow.swf.helpers import swf_identity
 from simpleflow.swf.task import ActivityTask, WorkflowTask, SignalTask
 from simpleflow.task import (
@@ -151,7 +150,6 @@ class Executor(executor.Executor):
 
     :ivar domain: domain
     :type domain: swf.models.domain.Domain
-    :ivar workflow: workflow
     :ivar task_list: task list
     :type task_list: Optional[str]
     :ivar repair_with: previous history to use for repairing
@@ -175,6 +173,7 @@ def __init__(self, domain, workflow_class, task_list=None, repair_with=None,
             self.force_activities = None
         self.reset()
 
+    # noinspection PyAttributeOutsideInit
     def reset(self):
         """
         Clears the state of the execution.
@@ -187,6 +186,8 @@ def reset(self):
         self._decisions = []
         self._tasks = TaskRegistry()
         self._idempotent_tasks_to_submit = set()
+        self._execution = None
+        self.current_priority = None
 
     def _make_task_id(self, a_task, *args, **kwargs):
         """

simpleflow/swf/process/decider/base.py

@@ -62,21 +62,22 @@ def __init__(self, workflow_executors, domain, task_list, nb_retries=3,
         :type  workflow_executors: list[simpleflow.swf.executor.Executor]
 
         """
+        print(ex._workflow for ex in workflow_executors)
         self._workflow_name = '{}'.format(','.join(
             [
-                ex.workflow.name for ex in workflow_executors
+                ex.workflow_class.name for ex in workflow_executors
                 ]))
 
         # Maps a workflow's name to its definition.
         # Used to dispatch a decision task to the corresponding workflow.
         self._workflow_executors = {
-            executor.workflow.name: executor for executor in workflow_executors
+            executor.workflow_class.name: executor for executor in workflow_executors
         }
 
         if task_list:
             self.task_list = task_list
         else:
-            self.task_list = workflow_executors[0].workflow.task_list
+            self.task_list = workflow_executors[0].workflow_class.task_list
             # If not passed explicitly, all executors must use the same task list
             # else it's probably a mistake so we raise an error.
             self._check_all_task_lists_identical()
@@ -106,7 +107,7 @@ def _check_all_domains_identical(self):
 
     def _check_all_task_lists_identical(self):
         for ex in self._workflow_executors.values():
-            if ex.workflow.task_list != self.task_list:
+            if ex.workflow_class.task_list != self.task_list:
                 raise ValueError(
                     'all workflows must have the same task list '
                     '"{}" unless you specify it explicitly'.format(