diff --git a/dvc/command/data_sync.py b/dvc/command/data_sync.py index ccf9f5906..4dd3af518 100644 --- a/dvc/command/data_sync.py +++ b/dvc/command/data_sync.py @@ -4,6 +4,7 @@ import argparse import logging from dvc.command.base import CmdBase, append_doc_link +from dvc.exceptions import DvcException logger = logging.getLogger(__name__) @@ -45,7 +46,7 @@ class CmdDataPull(CmdDataBase): force=self.args.force, recursive=self.args.recursive, ) - except Exception: + except DvcException: logger.exception("failed to pull data from the cloud") return 1 self.check_up_to_date(processed_files_count) @@ -65,7 +66,7 @@ class CmdDataPush(CmdDataBase): with_deps=self.args.with_deps, recursive=self.args.recursive, ) - except Exception: + except DvcException: logger.exception("failed to push data to the cloud") return 1 self.check_up_to_date(processed_files_count) @@ -85,7 +86,7 @@ class CmdDataFetch(CmdDataBase): with_deps=self.args.with_deps, recursive=self.args.recursive, ) - except Exception: + except DvcException: logger.exception("failed to fetch data from the cloud") return 1 self.check_up_to_date(processed_files_count) diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 12ef420b2..2d6acb12b 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -390,6 +390,7 @@ class RemoteBASE(object): if not hasattr(self, "_upload"): raise RemoteActionNotImplemented("upload", self.scheme) names = self._verify_path_args(to_infos, from_infos, names) + fails = 0 with self.transfer_context() as ctx: for from_info, to_info, name in zip(from_infos, to_infos, names): @@ -417,6 +418,7 @@ class RemoteBASE(object): no_progress_bar=no_progress_bar, ) except Exception: + fails += 1 msg = "failed to upload '{}' to '{}'" logger.exception(msg.format(from_info, to_info)) continue @@ -424,18 +426,21 @@ class RemoteBASE(object): if not no_progress_bar: progress.finish_target(name) + return fails + def download( self, from_infos, to_infos, - no_progress_bar=False, names=None, + no_progress_bar=False, resume=False, ): if not hasattr(self, "_download"): raise RemoteActionNotImplemented("download", self.scheme) names = self._verify_path_args(from_infos, to_infos, names) + fails = 0 with self.transfer_context() as ctx: for to_info, from_info, name in zip(to_infos, from_infos, names): @@ -473,6 +478,7 @@ class RemoteBASE(object): no_progress_bar=no_progress_bar, ) except Exception: + fails += 1 msg = "failed to download '{}' to '{}'" logger.exception(msg.format(from_info, to_info)) continue @@ -482,6 +488,8 @@ class RemoteBASE(object): if not no_progress_bar: progress.finish_target(name) + return fails + def remove(self, path_info): raise RemoteActionNotImplemented("remove", self.scheme) diff --git a/dvc/remote/http.py b/dvc/remote/http.py index 20498cbf3..3ecfe90a8 100644 --- a/dvc/remote/http.py +++ b/dvc/remote/http.py @@ -48,11 +48,12 @@ class RemoteHTTP(RemoteBASE): self, from_infos, to_infos, - no_progress_bar=False, names=None, + no_progress_bar=False, resume=False, ): names = self._verify_path_args(to_infos, from_infos, names) + fails = 0 for to_info, from_info, name in zip(to_infos, from_infos, names): if from_info.scheme != self.scheme: @@ -82,6 +83,7 @@ class RemoteHTTP(RemoteBASE): ) except Exception: + fails += 1 msg = "failed to download '{}'".format(from_info) logger.exception(msg) continue @@ -89,6 +91,8 @@ class RemoteHTTP(RemoteBASE): if not no_progress_bar: progress.finish_target(name) + return fails + def exists(self, path_infos): single_path = False diff --git a/dvc/remote/local/__init__.py b/dvc/remote/local/__init__.py index cd6ffd67b..898f34ab4 100644 --- a/dvc/remote/local/__init__.py +++ b/dvc/remote/local/__init__.py @@ -355,26 +355,16 @@ class RemoteLOCAL(RemoteBASE): info["status"] = status def _get_chunks(self, download, remote, status_info, status, jobs): - title = "Analysing status." - - progress.set_n_total(1) - total = len(status_info) - current = 0 - cache = [] path_infos = [] names = [] - for md5, info in status_info.items(): + for md5, info in progress( + status_info.items(), name="Analysing status" + ): if info["status"] == status: cache.append(self.checksum_to_path_info(md5)) path_infos.append(remote.checksum_to_path_info(md5)) names.append(info["name"]) - current += 1 - progress.update_target(title, current, total) - - progress.finish_target(title) - - progress.set_n_total(len(names)) if download: to_infos = cache @@ -383,12 +373,10 @@ class RemoteLOCAL(RemoteBASE): to_infos = path_infos from_infos = cache - return list( - zip( - to_chunks(from_infos, jobs), - to_chunks(to_infos, jobs), - to_chunks(names, jobs), - ) + return ( + to_chunks(from_infos, jobs), + to_chunks(to_infos, jobs), + to_chunks(names, jobs), ) def _process( @@ -399,11 +387,9 @@ class RemoteLOCAL(RemoteBASE): show_checksums=False, download=False, ): - msg = "Preparing to {} data {} '{}'" logger.info( - msg.format( - "download" if download else "upload", - "from" if download else "to", + "Preparing to {} '{}'".format( + "download data from" if download else "upload data to", remote.path_info, ) ) @@ -428,25 +414,22 @@ class RemoteLOCAL(RemoteBASE): chunks = self._get_chunks(download, remote, status_info, status, jobs) - if len(chunks) == 0: + if len(chunks[0]) == 0: return 0 if jobs > 1: - futures = [] with ThreadPoolExecutor(max_workers=jobs) as executor: - for from_infos, to_infos, names in chunks: - res = executor.submit( - func, from_infos, to_infos, names=names - ) - futures.append(res) - - for f in futures: - f.result() + fails = sum(executor.map(func, *chunks)) else: - for from_infos, to_infos, names in chunks: - func(from_infos, to_infos, names=names) + fails = sum(map(func, *chunks)) - return len(chunks) + if fails: + msg = "{} file(s) failed to {}" + raise DvcException( + msg.format(fails, "download" if download else "upload") + ) + + return len(chunks[0]) def push(self, checksum_infos, remote, jobs=None, show_checksums=False): return self._process( diff --git a/tests/func/test_remote.py b/tests/func/test_remote.py index 0a6f74dc0..da4a1de35 100644 --- a/tests/func/test_remote.py +++ b/tests/func/test_remote.py @@ -1,14 +1,17 @@ import os import configobj +import shutil from mock import patch from dvc.main import main from dvc.config import Config from dvc.remote.base import RemoteBASE +from dvc.remote import RemoteLOCAL from dvc.path_info import PathInfo from tests.basic_env import TestDvc from tests.func.test_data_cloud import get_local_storagepath +from .test_data_cloud import get_local_url class TestRemote(TestDvc): @@ -205,3 +208,41 @@ def test_dir_checksum_should_be_key_order_agnostic(dvc_repo): checksum2 = dvc_repo.cache.local.get_dir_checksum(path_info) assert checksum1 == checksum2 + + +def test_partial_push_n_pull(dvc_repo, repo_dir, caplog): + assert main(["remote", "add", "-d", "upstream", get_local_url()]) == 0 + # Recreate the repo to reread config + repo = dvc_repo.__class__(dvc_repo.root_dir) + remote = repo.cloud.get_remote("upstream") + + foo = repo.add(repo_dir.FOO)[0].outs[0] + bar = repo.add(repo_dir.BAR)[0].outs[0] + + # Faulty upload version, failing on foo + original = RemoteLOCAL._upload + + def unreliable_upload(self, from_file, to_info, name=None, **kwargs): + if name == "foo": + raise Exception("stop foo") + return original(self, from_file, to_info, name, **kwargs) + + with patch.object(RemoteLOCAL, "_upload", unreliable_upload): + assert main(["push"]) == 1 + assert str(get_last_exc(caplog)) == "1 file(s) failed to upload" + + assert not remote.exists(remote.checksum_to_path_info(foo.checksum)) + assert remote.exists(remote.checksum_to_path_info(bar.checksum)) + + # Push everything and delete local cache + assert main(["push"]) == 0 + shutil.rmtree(repo.cache.local.cache_dir) + + with patch.object(RemoteLOCAL, "_download", side_effect=Exception): + assert main(["pull"]) == 1 + assert str(get_last_exc(caplog)) == "2 file(s) failed to download" + + +def get_last_exc(caplog): + _, exc, _ = caplog.records[-1].exc_info + return exc