Merge pull request #883 from efiop/master

add `dvc pipeline show`

dvc/cli.py

@@ -23,6 +23,7 @@ from dvc.command.metrics import CmdMetricsShow, CmdMetricsAdd, CmdMetricsRemove
 from dvc.command.install import CmdInstall
 from dvc.command.root import CmdRoot
 from dvc.command.lock import CmdLock, CmdUnlock
+from dvc.command.pipeline import CmdPipelineShow
 from dvc.stage import Stage
 from dvc import VERSION
 
@@ -527,6 +528,33 @@ def parse_args(argv=None):
                         help='DVC files.')
     unlock_parser.set_defaults(func=CmdUnlock)
 
+    # Pipeline
+    pipeline_parser = subparsers.add_parser(
+                        'pipeline',
+                        parents=[parent_parser],
+                        help='Pipeline')
+
+    pipeline_subparsers = pipeline_parser.add_subparsers(
+                        dest='cmd',
+                        help='Use dvc pipeline CMD --help for command-specific help')
+
+    _fix_subparsers(pipeline_subparsers)
+
+    pipeline_show_parser = pipeline_subparsers.add_parser(
+                        'show',
+                        parents=[parent_parser],
+                        help='Show pipeline')
+    pipeline_show_parser.add_argument('-c',
+                        '--commands',
+                        action='store_true',
+                        default=False,
+                        help='Print commands instead of paths to DVC files.')
+    pipeline_show_parser.add_argument(
+                        'targets',
+                        nargs='+',
+                        help='DVC files.')
+    pipeline_show_parser.set_defaults(func=CmdPipelineShow)
+                        
 
     args = parser.parse_args(argv)
 

dvc/command/pipeline.py

@@ -0,0 +1,31 @@
+import os
+import networkx
+
+from dvc.exceptions import DvcException
+from dvc.command.common.base import CmdBase
+from dvc.project import NotDvcFileError
+
+
+class CmdPipelineShow(CmdBase):
+    def _show(self, target, commands):
+        G = self.project.graph()
+        stages = networkx.get_node_attributes(G, 'stage')
+        node = os.path.relpath(os.path.abspath(target), self.project.root_dir)
+        if node not in stages:
+            raise NotDvcFileError(node)
+
+        for n in networkx.dfs_postorder_nodes(G, node):
+            line = n
+            if commands:
+                line = stages[n].cmd
+            self.project.logger.info(line)
+
+    def run(self, unlock=False):
+        for target in self.args.targets:
+            try:
+                self._show(target, self.args.commands)
+            except DvcException as ex:
+                msg = 'Failed to show pipeline for \'{}\''.format(target)
+                self.project.logger.error(msg, ex)
+                return 1
+        return 0

dvc/project.py

@@ -227,19 +227,20 @@ class Project(object):
         return [stage]
 
     def reproduce(self, target, recursive=True, force=False):
-        stages = nx.get_node_attributes(self.graph(), 'stage')
+        G = self.graph()
+        stages = nx.get_node_attributes(G, 'stage')
         node = os.path.relpath(os.path.abspath(target), self.root_dir)
         if node not in stages:
             raise NotDvcFileError(target)
 
         if recursive:
-            return self._reproduce_stages(stages, node, force)
+            return self._reproduce_stages(G, stages, node, force)
 
         return self._reproduce_stage(stages, node, force)
 
-    def _reproduce_stages(self, stages, node, force):
+    def _reproduce_stages(self, G, stages, node, force):
         result = []
-        for n in nx.dfs_postorder_nodes(self.graph(), node):
+        for n in nx.dfs_postorder_nodes(G, node):
             try:
                 result += self._reproduce_stage(stages, n, force)
             except Exception as ex:

tests/test_pipeline.py

@@ -0,0 +1,21 @@
+from dvc.main import main
+
+from tests.test_repro import TestRepro
+
+
+class TestPipelineShow(TestRepro):
+    def test(self):
+        ret = main(['pipeline', 'show', self.file1_stage])
+        self.assertEqual(ret, 0)
+
+    def test_commands(self):
+        ret = main(['pipeline', 'show', self.file1_stage, '--commands'])
+        self.assertEqual(ret, 0)        
+
+    def test_not_dvc_file(self):
+        ret = main(['pipeline', 'show', self.file1])
+        self.assertNotEqual(ret, 0)
+
+    def test_non_existing(self):
+        ret = main(['pipeline', 'show', 'non-existing'])
+        self.assertNotEqual(ret, 0)