dvc: batch push/pull/fetch/status

This will speed up cache collection and then run single cloud op for all
collected cache.

Fixes #2308.

dvc/command/data_sync.py

@@ -11,32 +11,17 @@ logger = logging.getLogger(__name__)
 
 
 class CmdDataBase(CmdBase):
-    UP_TO_DATE_MSG = "Everything is up to date."
-
-    def do_run(self, target):
-        pass
-
-    def run(self):
-        if not self.args.targets:
-            return self.do_run()
-
-        ret = 0
-        for target in self.args.targets:
-            if self.do_run(target):
-                ret = 1
-        return ret
-
     @classmethod
     def check_up_to_date(cls, processed_files_count):
         if processed_files_count == 0:
-            logger.info(cls.UP_TO_DATE_MSG)
+            logger.info("Everything is up to date.")
 
 
 class CmdDataPull(CmdDataBase):
-    def do_run(self, target=None):
+    def run(self):
         try:
             processed_files_count = self.repo.pull(
-                target=target,
+                targets=self.args.targets,
                 jobs=self.args.jobs,
                 remote=self.args.remote,
                 show_checksums=self.args.show_checksums,
@@ -54,10 +39,10 @@ class CmdDataPull(CmdDataBase):
 
 
 class CmdDataPush(CmdDataBase):
-    def do_run(self, target=None):
+    def run(self):
         try:
             processed_files_count = self.repo.push(
-                target=target,
+                targets=self.args.targets,
                 jobs=self.args.jobs,
                 remote=self.args.remote,
                 show_checksums=self.args.show_checksums,
@@ -74,10 +59,10 @@ class CmdDataPush(CmdDataBase):
 
 
 class CmdDataFetch(CmdDataBase):
-    def do_run(self, target=None):
+    def run(self):
         try:
             processed_files_count = self.repo.fetch(
-                target=target,
+                targets=self.args.targets,
                 jobs=self.args.jobs,
                 remote=self.args.remote,
                 show_checksums=self.args.show_checksums,

dvc/command/status.py

@@ -40,11 +40,11 @@ class CmdDataStatus(CmdDataBase):
                 logger.info("{}{}:".format(ind, key))
                 self._show(value, indent + 1)
 
-    def do_run(self, target=None):
+    def run(self):
         indent = 1 if self.args.cloud else 0
         try:
             st = self.repo.status(
-                target=target,
+                targets=self.args.targets,
                 jobs=self.args.jobs,
                 cloud=self.args.cloud,
                 show_checksums=self.args.show_checksums,

dvc/repo/__init__.py

@@ -180,7 +180,7 @@ class Repo(object):
 
     def used_cache(
         self,
-        target=None,
+        targets=None,
         all_branches=False,
         active=True,
         with_deps=False,
@@ -215,18 +215,22 @@ class Repo(object):
         for branch in self.brancher(
             all_branches=all_branches, all_tags=all_tags
         ):
-            if target:
-                if recursive and os.path.isdir(target):
-                    stages = self.stages(target)
-                else:
-                    stages = self.collect(target, with_deps=with_deps)
+            if targets:
+                stages = []
+                for target in targets:
+                    if recursive and os.path.isdir(target):
+                        stages.extend(self.stages(target))
+                    else:
+                        stages.extend(
+                            self.collect(target, with_deps=with_deps)
+                        )
             elif active:
                 stages = self.active_stages()
             else:
                 stages = self.stages()
 
             for stage in stages:
-                if active and not target and stage.locked:
+                if active and not targets and stage.locked:
                     logger.warning(
                         "DVC-file '{path}' is locked. Its dependencies are"
                         " not going to be pushed/pulled/fetched.".format(

dvc/repo/fetch.py

@@ -3,7 +3,7 @@ from __future__ import unicode_literals
 
 def fetch(
     self,
-    target=None,
+    targets=None,
     jobs=None,
     remote=None,
     all_branches=False,
@@ -14,7 +14,7 @@ def fetch(
 ):
     with self.state:
         used = self.used_cache(
-            target,
+            targets,
             all_branches=all_branches,
             all_tags=all_tags,
             with_deps=with_deps,

dvc/repo/gc.py

@@ -35,7 +35,7 @@ def _load_all_used_cache(
     for repo in repos:
         with repo.state:
             repo_clist = repo.used_cache(
-                target=None,
+                targets=None,
                 all_branches=all_branches,
                 active=False,
                 with_deps=with_deps,

dvc/repo/pull.py

@@ -3,7 +3,7 @@ from __future__ import unicode_literals
 
 def pull(
     self,
-    target=None,
+    targets=None,
     jobs=None,
     remote=None,
     all_branches=False,
@@ -14,7 +14,7 @@ def pull(
     recursive=False,
 ):
     processed_files_count = self.fetch(
-        target,
+        targets,
         jobs,
         remote=remote,
         all_branches=all_branches,
@@ -23,7 +23,11 @@ def pull(
         with_deps=with_deps,
         recursive=recursive,
     )
-    self.checkout(
-        target=target, with_deps=with_deps, force=force, recursive=recursive
-    )
+    for target in targets or [None]:
+        self.checkout(
+            target=target,
+            with_deps=with_deps,
+            force=force,
+            recursive=recursive,
+        )
     return processed_files_count

dvc/repo/push.py

@@ -3,7 +3,7 @@ from __future__ import unicode_literals
 
 def push(
     self,
-    target=None,
+    targets=None,
     jobs=None,
     remote=None,
     all_branches=False,
@@ -14,7 +14,7 @@ def push(
 ):
     with self.state:
         used = self.used_cache(
-            target,
+            targets,
             all_branches=all_branches,
             all_tags=all_tags,
             with_deps=with_deps,

dvc/repo/status.py

@@ -1,15 +1,19 @@
 from __future__ import unicode_literals
 
 import logging
+from funcy.py3 import cat
 
 
 logger = logging.getLogger(__name__)
 
 
-def _local_status(self, target=None, with_deps=False):
+def _local_status(self, targets=None, with_deps=False):
     status = {}
 
-    stages = self.collect(target, with_deps=with_deps)
+    if targets:
+        stages = cat(self.collect(t, with_deps=with_deps) for t in targets)
+    else:
+        stages = self.collect(None, with_deps=with_deps)
 
     for stage in stages:
         if stage.locked:
@@ -27,7 +31,7 @@ def _local_status(self, target=None, with_deps=False):
 
 def _cloud_status(
     self,
-    target=None,
+    targets=None,
     jobs=None,
     remote=None,
     show_checksums=False,
@@ -64,7 +68,7 @@ def _cloud_status(
     import dvc.remote.base as cloud
 
     used = self.used_cache(
-        target,
+        targets,
         all_branches=all_branches,
         all_tags=all_tags,
         with_deps=with_deps,
@@ -92,7 +96,7 @@ def _cloud_status(
 
 def status(
     self,
-    target=None,
+    targets=None,
     jobs=None,
     cloud=False,
     remote=None,
@@ -105,7 +109,7 @@ def status(
         if cloud or remote:
             return _cloud_status(
                 self,
-                target,
+                targets,
                 jobs,
                 remote=remote,
                 show_checksums=show_checksums,
@@ -113,4 +117,4 @@ def status(
                 with_deps=with_deps,
                 all_tags=all_tags,
             )
-        return _local_status(self, target, with_deps=with_deps)
+        return _local_status(self, targets, with_deps=with_deps)

tests/func/test_commit.py

@@ -37,7 +37,7 @@ def test_commit_force(dvc_repo, repo_dir):
 
     dvc_repo.commit(stage.path, force=True)
 
-    assert dvc_repo.status(stage.path) == {}
+    assert dvc_repo.status([stage.path]) == {}
 
 
 def test_commit_with_deps(dvc_repo, repo_dir):

tests/func/test_repro.py

@@ -898,13 +898,13 @@ class TestReproExternalBase(TestDvc):
 
         self.assertTrue(os.path.exists("import"))
         self.assertTrue(filecmp.cmp("import", self.FOO, shallow=False))
-        self.assertEqual(self.dvc.status(import_stage.path), {})
+        self.assertEqual(self.dvc.status([import_stage.path]), {})
         self.check_already_cached(import_stage)
 
         import_remote_stage = self.dvc.imp_url(
             out_foo_path, out_foo_path + "_imported"
         )
-        self.assertEqual(self.dvc.status(import_remote_stage.path), {})
+        self.assertEqual(self.dvc.status([import_remote_stage.path]), {})
 
         cmd_stage = self.dvc.run(
             outs=[out_bar_path],
@@ -912,7 +912,7 @@ class TestReproExternalBase(TestDvc):
             cmd=self.cmd(foo_path, bar_path),
         )
 
-        self.assertEqual(self.dvc.status(cmd_stage.path), {})
+        self.assertEqual(self.dvc.status([cmd_stage.path]), {})
         self.assertEqual(self.dvc.status(), {})
         self.check_already_cached(cmd_stage)
 
@@ -923,24 +923,24 @@ class TestReproExternalBase(TestDvc):
         self.dvc.update(import_stage.path)
         self.assertTrue(os.path.exists("import"))
         self.assertTrue(filecmp.cmp("import", self.BAR, shallow=False))
-        self.assertEqual(self.dvc.status(import_stage.path), {})
+        self.assertEqual(self.dvc.status([import_stage.path]), {})
 
         self.dvc.update(import_remote_stage.path)
-        self.assertEqual(self.dvc.status(import_remote_stage.path), {})
+        self.assertEqual(self.dvc.status([import_remote_stage.path]), {})
 
         stages = self.dvc.reproduce(cmd_stage.path)
         self.assertEqual(len(stages), 1)
-        self.assertEqual(self.dvc.status(cmd_stage.path), {})
+        self.assertEqual(self.dvc.status([cmd_stage.path]), {})
 
         self.assertEqual(self.dvc.status(), {})
         self.dvc.gc()
         self.assertEqual(self.dvc.status(), {})
 
         self.dvc.remove(cmd_stage.path, outs_only=True)
-        self.assertNotEqual(self.dvc.status(cmd_stage.path), {})
+        self.assertNotEqual(self.dvc.status([cmd_stage.path]), {})
 
         self.dvc.checkout(cmd_stage.path, force=True)
-        self.assertEqual(self.dvc.status(cmd_stage.path), {})
+        self.assertEqual(self.dvc.status([cmd_stage.path]), {})
 
 
 @pytest.mark.skipif(os.name == "nt", reason="temporarily disabled on windows")
@@ -1207,7 +1207,7 @@ class TestReproExternalHTTP(TestReproExternalBase):
             shutil.move(self.local_cache, cache_id)
             self.assertFalse(os.path.exists(self.local_cache))
 
-            self.dvc.pull(import_stage.path, remote="mycache")
+            self.dvc.pull([import_stage.path], remote="mycache")
 
             self.assertTrue(os.path.exists(import_output))
 

tests/func/test_update.py

@@ -34,9 +34,9 @@ def test_update_import(dvc_repo, erepo):
 
     os.chdir(saved_dir)
 
-    assert dvc_repo.status(stage.path) == {}
+    assert dvc_repo.status([stage.path]) == {}
     dvc_repo.update(stage.path)
-    assert dvc_repo.status(stage.path) == {}
+    assert dvc_repo.status([stage.path]) == {}
 
     assert os.path.exists(dst)
     assert os.path.isfile(dst)
@@ -60,9 +60,9 @@ def test_update_import_url(repo_dir, dvc_repo):
     os.unlink(src)
     shutil.copyfile(repo_dir.BAR, src)
 
-    assert dvc_repo.status(stage.path) == {}
+    assert dvc_repo.status([stage.path]) == {}
     dvc_repo.update(stage.path)
-    assert dvc_repo.status(stage.path) == {}
+    assert dvc_repo.status([stage.path]) == {}
 
     assert os.path.exists(dst)
     assert os.path.isfile(dst)