Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-02 07:18:25 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2342 from Suor/mass-ops

Browse Source 
dvc: batch push/pull/fetch/status
This commit is contained in:
Ruslan Kuprieiev
2019-08-01 20:57:31 +03:00
committed by GitHub
parent b1f38d8e72 50b5b87304
commit 9f173fa174
11 changed files with 59 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/command/data_sync.py
+7 -22
View File
@@ -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
+2 -2
View File
@@ -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
+11 -7
View File
@@ -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
+2 -2
View File
@@ -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
+1 -1
View File
@@ -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
+9 -5
View File
@@ -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
+2 -2
View File
@@ -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
+11 -7
View File
@@ -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
+1 -1
View File
@@ -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
+9 -9
View File
@@ -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
+4 -4
View File
@@ -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)