Merge branch 'master' of github.com:yahoo/TaskFlow into sql

3 files changed, 64 insertions, 41 deletions

diff --git a/taskflow/patterns/graph_workflow.py b/taskflow/patterns/graph_flow.py
index 8f8f05e..b9943da 100644
--- a/taskflow/patterns/graph_workflow.py
+++ b/taskflow/patterns/graph_flow.py
@@ -25,18 +25,18 @@ from networkx.algorithms import dag
 from networkx.classes import digraph
 
 from taskflow import exceptions as exc
-from taskflow.patterns import ordered_workflow
+from taskflow.patterns import ordered_flow
 
 LOG = logging.getLogger(__name__)
 
 
-class Workflow(ordered_workflow.Workflow):
-    """A workflow which will analyze the attached tasks input requirements and
+class Flow(ordered_flow.Flow):
+    """A flow which will analyze the attached tasks input requirements and
     determine who provides said input and order the task so that said providing
     task will be ran before."""
 
     def __init__(self, name, tolerant=False, parents=None):
-        super(Workflow, self).__init__(name, tolerant, parents)
+        super(Flow, self).__init__(name, tolerant, parents)
         self._graph = digraph.DiGraph()
         self._connected = False
 
@@ -49,9 +49,20 @@ class Workflow(ordered_workflow.Workflow):
         self._graph.add_node(task)
         self._connected = False
 
+    def _fetch_task_inputs(self, task):
+        task_needs = task.requires()
+        if not task_needs:
+            return None
+        inputs = {}
+        for (_who, there_result) in self.results:
+            for n in task_needs:
+                if there_result and n in there_result:
+                    inputs[n] = there_result[n]
+        return inputs
+
     def run(self, context, *args, **kwargs):
         self.connect()
-        return super(Workflow, self).run(context, *args, **kwargs)
+        return super(Flow, self).run(context, *args, **kwargs)
 
     def order(self):
         self.connect()
@@ -60,27 +71,39 @@ class Workflow(ordered_workflow.Workflow):
         except g_exc.NetworkXUnfeasible:
             raise exc.InvalidStateException("Unable to correctly determine "
                                             "the path through the provided "
-                                            "workflow which will satisfy the "
+                                            "flow which will satisfy the "
                                             "tasks needed inputs and outputs.")
 
     def connect(self):
-        """Connects the edges of the graph together."""
-        if self._connected:
+        """Connects the nodes & edges of the graph together."""
+        if self._connected or len(self._graph) == 0:
             return
+
         provides_what = defaultdict(list)
         requires_what = defaultdict(list)
         for t in self._graph.nodes_iter():
-            for r in t.requires:
+            for r in t.requires():
                 requires_what[r].append(t)
-            for p in t.provides:
+            for p in t.provides():
                 provides_what[p].append(t)
-        for (i_want, n) in requires_what.items():
-            if i_want not in provides_what:
-                raise exc.InvalidStateException("Task %s requires input %s "
-                                                "but no other task produces "
-                                                "said output" % (n, i_want))
-            for p in provides_what[i_want]:
+
+        for (want_what, who_wants) in requires_what.items():
+            who_provided = 0
+            for p in provides_what.get(want_what, []):
                 # P produces for N so thats why we link P->N and not N->P
-                self._graph.add_edge(p, n)
+                for n in who_wants:
+                    if p is n:
+                        # No self-referencing allowed.
+                        continue
+                    why = {
+                        want_what: True,
+                    }
+                    self._graph.add_edge(p, n, why)
+                    who_provided += 1
+            if not who_provided:
+                raise exc.InvalidStateException("Task/s %s requires input %s "
+                                                "but no other task produces "
+                                                "said output." % (who_wants,
+                                                                  want_what))
 
         self._connected = True
diff --git a/taskflow/patterns/linear_workflow.py b/taskflow/patterns/linear_flow.py
index 5844797..9283c05 100644
--- a/taskflow/patterns/linear_workflow.py
+++ b/taskflow/patterns/linear_flow.py
@@ -16,15 +16,15 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-from taskflow.patterns import ordered_workflow
+from taskflow.patterns import ordered_flow
 
 
-class Workflow(ordered_workflow.Workflow):
+class Flow(ordered_flow.Flow):
     """A linear chain of *independent* tasks that can be applied as one unit or
        rolled back as one unit."""
 
     def __init__(self, name, tolerant=False, parents=None):
-        super(Workflow, self).__init__(name, tolerant, parents)
+        super(Flow, self).__init__(name, tolerant, parents)
         self._tasks = []
 
     def add(self, task):
diff --git a/taskflow/patterns/ordered_workflow.py b/taskflow/patterns/ordered_flow.py
index 24813d8..885debc 100644
--- a/taskflow/patterns/ordered_workflow.py
+++ b/taskflow/patterns/ordered_flow.py
@@ -28,10 +28,10 @@ from taskflow import states
 LOG = logging.getLogger(__name__)
 
 
-class Workflow(object):
+class Flow(object):
     """A set tasks that can be applied as one unit or rolled back as one
-    unit using an ordered arrangements of said tasks where reversion can be
-    handled by reversing through the tasks applied."""
+    unit using an ordered arrangements of said tasks where reversion is by
+    default handled by reversing through the tasks applied."""
 
     __metaclass__ = abc.ABCMeta
 
@@ -43,8 +43,8 @@ class Workflow(object):
         # If this chain can ignore individual task reversion failure then this
         # should be set to true, instead of the default value of false.
         self.tolerant = tolerant
-        # If this workflow has a parent workflow/s which need to be reverted if
-        # this workflow fails then please include them here to allow this child
+        # If this flow has a parent flow/s which need to be reverted if
+        # this flow fails then please include them here to allow this child
         # to call the parents...
         self.parents = parents
         # This should be a functor that returns whether a given task has
@@ -73,11 +73,11 @@ class Workflow(object):
 
     @abc.abstractmethod
     def add(self, task):
-        """Adds a given task to this workflow."""
+        """Adds a given task to this flow."""
         raise NotImplementedError()
 
     def __str__(self):
-        return "Workflow: %s" % (self.name)
+        return "Flow: %s" % (self.name)
 
     @abc.abstractmethod
     def order(self):
@@ -93,17 +93,17 @@ class Workflow(object):
     def _perform_reconcilation(self, context, task, excp):
         # Attempt to reconcile the given exception that occured while applying
         # the given task and either reconcile said task and its associated
-        # failure, so that the workflow can continue or abort and perform
+        # failure, so that the flow can continue or abort and perform
         # some type of undo of the tasks already completed.
-        try:
-            self._change_state(context, states.REVERTING)
-        except Exception:
-            LOG.exception("Dropping exception catched when"
-                          " changing state to reverting while performing"
-                          " reconcilation on a tasks exception.")
         cause = exc.TaskException(task, self, excp)
         with excutils.save_and_reraise_exception():
             try:
+                self._change_state(context, states.REVERTING)
+            except Exception:
+                LOG.exception("Dropping exception catched when"
+                              " changing state to reverting while performing"
+                              " reconcilation on a tasks exception.")
+            try:
                 self._on_task_error(context, task)
             except Exception:
                 LOG.exception("Dropping exception catched when"
@@ -111,7 +111,7 @@ class Workflow(object):
                               " exception.")
             # The default strategy will be to rollback all the contained
             # tasks by calling there reverting methods, and then calling
-            # any parent workflows rollbacks (and so-on).
+            # any parent flows rollbacks (and so-on).
             try:
                 self.rollback(context, cause)
             finally:
@@ -124,7 +124,7 @@ class Workflow(object):
 
     def run(self, context, *args, **kwargs):
         if self.state != states.PENDING:
-            raise exc.InvalidStateException("Unable to run workflow when "
+            raise exc.InvalidStateException("Unable to run flow when "
                                             "in state %s" % (self.state))
 
         if self.result_fetcher:
@@ -233,10 +233,10 @@ class Workflow(object):
     def rollback(self, context, cause):
         # Performs basic task by task rollback by going through the reverse
         # order that tasks have finished and asking said task to undo whatever
-        # it has done. If this workflow has any parent workflows then they will
+        # it has done. If this flow has any parent flows then they will
         # also be called to rollback any tasks said parents contain.
         #
-        # Note(harlowja): if a workflow can more simply revert a whole set of
+        # Note(harlowja): if a flow can more simply revert a whole set of
         # tasks via a simpler command then it can override this method to
         # accomplish that.
         #
@@ -253,14 +253,14 @@ class Workflow(object):
                 if not self.tolerant:
                     log_f = LOG.exception
                 msg = ("Failed rolling back stage %(index)s (%(task)s)"
-                       " of workflow %(workflow)s, due to inner exception.")
-                log_f(msg % {'index': (i + 1), 'task': task, 'workflow': self})
+                       " of flow %(flow)s, due to inner exception.")
+                log_f(msg % {'index': (i + 1), 'task': task, 'flow': self})
                 if not self.tolerant:
                     # NOTE(harlowja): LOG a msg AND re-raise the exception if
                     # the chain does not tolerate exceptions happening in the
                     # rollback method.
                     raise
         if self.parents:
-            # Rollback any parents workflows if they exist...
+            # Rollback any parents flows if they exist...
             for p in self.parents:
                 p.rollback(context, cause)