diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 9d67ce099..8eb18829e 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -4,6 +4,7 @@ import errno import posixpath from multiprocessing import cpu_count +import dvc.prompt as prompt from dvc.config import Config from dvc.logger import logger from dvc.exceptions import DvcException @@ -105,9 +106,6 @@ class RemoteBase(object): def save(self, path_info): raise NotImplementedError - def checkout(self, path_info, checksum_info, force=False): - raise NotImplementedError - def download(self, from_infos, to_infos, no_progress_bar=False, name=None): raise NotImplementedError @@ -120,6 +118,9 @@ class RemoteBase(object): def move(self, path_info): raise NotImplementedError + def copy(self, from_info, to_info): + raise NotImplementedError + def _makedirs(self, fname): dname = os.path.dirname(fname) @@ -200,8 +201,13 @@ class RemoteBase(object): def changed_cache(self, checksum): cache = self.checksum_to_path_info(checksum) + expected = {self.PARAM_CHECKSUM: checksum} + actual = self.save_info(cache) - if {self.PARAM_CHECKSUM: checksum} != self.save_info(cache): + logger.debug("Cache '{}' actual '{}'.".format(str(expected), + str(actual))) + + if expected != actual: if self.exists(cache): msg = 'Corrupted cache file {}' logger.warn(msg.format(str(cache))) @@ -227,3 +233,65 @@ class RemoteBase(object): if checksum == existing: ret[i] = True return ret + + def already_cached(self, path_info): + current = self.save_info(path_info)[self.PARAM_CHECKSUM] + + if not current: + return False + + return not self.changed_cache(current) + + def safe_remove(self, path_info, force=False): + if not self.exists(path_info): + return + + if not force and not self.already_cached(path_info): + msg = ( + 'File "{}" is going to be removed. ' + 'Are you sure you want to proceed?' + .format(str(path_info)) + ) + + if not prompt.confirm(msg): + raise DvcException("Unable to remove {} without a confirmation" + " from the user. Use '-f' to force." + .format(str(path_info))) + + self.remove(path_info) + + def do_checkout(self, path_info, checksum, force=False): + if self.exists(path_info): + msg = "Data '{}' exists. Removing before checkout." + logger.warn(msg.format(str(path_info))) + self.safe_remove(path_info, force=force) + + from_info = self.checksum_to_path_info(checksum) + self.copy(from_info, path_info) + + def checkout(self, path_info, checksum_info, force=False): + scheme = path_info['scheme'] + if scheme not in ['', 'local'] and scheme != self.scheme: + raise NotImplementedError + + checksum = checksum_info.get(self.PARAM_CHECKSUM) + if not checksum: + msg = "No checksum info for '{}'." + logger.info(msg.format(str(path_info))) + return + + if not self.changed(path_info, checksum_info): + msg = "Data '{}' didn't change." + logger.info(msg.format(str(path_info))) + return + + if self.changed_cache(checksum): + msg = "Cache '{}' not found. File '{}' won't be created." + logger.warn(msg.format(checksum, str(path_info))) + self.safe_remove(path_info, force=force) + return + + msg = "Checking out '{}' with cache '{}'." + logger.info(msg.format(str(path_info), checksum)) + + self.do_checkout(path_info, checksum, force=force) diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index 387221773..531e4e7e8 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -67,12 +67,6 @@ class RemoteGS(RemoteBase): return {self.PARAM_CHECKSUM: self.get_md5(path_info['bucket'], path_info['path'])} - @staticmethod - def to_string(path_info): - return "{}://{}/{}".format(path_info['scheme'], - path_info['bucket'], - path_info['path']) - def changed(self, path_info, checksum_info): if not self.exists(path_info): return True @@ -86,7 +80,7 @@ class RemoteGS(RemoteBase): return checksum_info != self.save_info(path_info) - def _copy(self, from_info, to_info, gs=None): + def copy(self, from_info, to_info, gs=None): gs = gs if gs else self.gs blob = gs.bucket(from_info['bucket']).get_blob(from_info['path']) @@ -107,42 +101,10 @@ class RemoteGS(RemoteBase): path = self.checksum_to_path(md5) to_info = {'scheme': 'gs', 'bucket': self.bucket, 'path': path} - self._copy(path_info, to_info) + self.copy(path_info, to_info) return {self.PARAM_CHECKSUM: md5} - def checkout(self, path_info, checksum_info): - if path_info['scheme'] != 'gs': - raise NotImplementedError - - md5 = checksum_info.get(self.PARAM_CHECKSUM, None) - if not md5: - return - - if not self.changed(path_info, checksum_info): - msg = "Data '{}' didn't change." - logger.info(msg.format(self.to_string(path_info))) - return - - if self.changed_cache(md5): - msg = "Cache '{}' not found. File '{}' won't be created." - logger.warn(msg.format(md5, self.to_string(path_info))) - return - - if self.exists(path_info): - msg = "Data '{}' exists. Removing before checkout." - logger.warn(msg.format(self.to_string(path_info))) - self.remove(path_info) - return - - msg = "Checking out '{}' with cache '{}'." - logger.info(msg.format(self.to_string(path_info), md5)) - - path = self.checksum_to_path(md5) - from_info = {'scheme': 'gs', 'bucket': self.bucket, 'path': path} - - self._copy(from_info, path_info) - def remove(self, path_info): if path_info['scheme'] != 'gs': raise NotImplementedError @@ -218,7 +180,7 @@ class RemoteGS(RemoteBase): raise NotImplementedError if to_info['scheme'] == 'gs': - self._copy(from_info, to_info, gs=gs) + self.copy(from_info, to_info, gs=gs) continue if to_info['scheme'] != 'local': diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py index 5eca0ba1c..dc114f894 100644 --- a/dvc/remote/hdfs.py +++ b/dvc/remote/hdfs.py @@ -65,7 +65,7 @@ class RemoteHDFS(RemoteBase): user=path_info['user']) return self._group(regex, stdout, 'checksum') - def cp(self, from_info, to_info): + def copy(self, from_info, to_info): dname = posixpath.dirname(to_info['path']) self.hadoop_fs('mkdir -p {}'.format(dname), user=to_info['user']) self.hadoop_fs('cp -f {} {}'.format(from_info['path'], @@ -84,11 +84,6 @@ class RemoteHDFS(RemoteBase): return {self.PARAM_CHECKSUM: self.checksum(path_info)} - @staticmethod - def to_string(path_info): - return "{}://{}".format(path_info['scheme'], - path_info['path']) - def changed(self, path_info, checksum_info): if not self.exists(path_info): return True @@ -112,44 +107,10 @@ class RemoteHDFS(RemoteBase): dest = path_info.copy() dest['path'] = self.checksum_to_path(checksum) - self.cp(path_info, dest) + self.copy(path_info, dest) return {self.PARAM_CHECKSUM: checksum} - def checkout(self, path_info, checksum_info): - if path_info['scheme'] != 'hdfs': - raise NotImplementedError - - assert path_info.get('path') - - checksum = checksum_info.get(self.PARAM_CHECKSUM, None) - if not checksum: - return - - if not self.changed(path_info, checksum_info): - msg = "Data '{}' didn't change." - logger.info(msg.format(self.to_string(path_info))) - return - - if self.changed_cache(checksum): - msg = "Cache '{}' not found. File '{}' won't be created." - logger.warn(msg.format(checksum, self.to_string(path_info))) - return - - if self.exists(path_info): - msg = "Data '{}' exists. Removing before checkout." - logger.warn(msg.format(self.to_string(path_info))) - self.remove(path_info) - return - - msg = "Checking out '{}' with cache '{}'." - logger.info(msg.format(self.to_string(path_info), checksum)) - - src = path_info.copy() - src['path'] = self.checksum_to_path(checksum) - - self.cp(src, path_info) - def remove(self, path_info): if path_info['scheme'] != 'hdfs': raise NotImplementedError @@ -199,7 +160,7 @@ class RemoteHDFS(RemoteBase): raise NotImplementedError if to_info['scheme'] == 'hdfs': - self.cp(from_info, to_info) + self.copy(from_info, to_info) continue if to_info['scheme'] != 'local': diff --git a/dvc/remote/local.py b/dvc/remote/local.py index ab9604352..5bce37e1a 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -7,7 +7,6 @@ import shutil import posixpath from operator import itemgetter -import dvc.prompt as prompt from dvc.system import System from dvc.remote.base import RemoteBase, STATUS_MAP from dvc.logger import logger @@ -270,36 +269,14 @@ class RemoteLOCAL(RemoteBase): def is_dir_cache(cls, cache): return cache.endswith(cls.MD5_DIR_SUFFIX) - def checkout(self, path_info, checksum_info, force=False): + def do_checkout(self, path_info, checksum, force=False): path = path_info['path'] - md5 = checksum_info.get(self.PARAM_CHECKSUM) + md5 = checksum cache = self.get(md5) - if not cache: - msg = 'No cache info for \'{}\'. Skipping checkout.' - logger.warn(msg.format(os.path.relpath(path))) - return - - if not self.changed(path_info, checksum_info): - msg = "Data '{}' didn't change." - logger.info(msg.format(os.path.relpath(path))) - return - - if self.changed_cache(md5): - msg = u'Cache \'{}\' not found. File \'{}\' won\'t be created.' - logger.warn(msg.format(md5, os.path.relpath(path))) - remove(path) - return - - msg = u'Checking out \'{}\' with cache \'{}\'.' - logger.info(msg.format(os.path.relpath(path), md5)) - if not self.is_dir_cache(cache): if os.path.exists(path): - if force or self._already_cached(path): - remove(path) - else: - self._safe_remove(path) + self.safe_remove(path_info, force=force) self.link(cache, path) self.state.update_link(path) @@ -329,10 +306,7 @@ class RemoteLOCAL(RemoteBase): if self.changed(entry_info, entry_checksum_info): if os.path.exists(p): - if force or self._already_cached(p): - remove(p) - else: - self._safe_remove(p) + self.safe_remove(entry_info, force=force) self.link(c, p) @@ -346,8 +320,10 @@ class RemoteLOCAL(RemoteBase): if bar: progress.finish_target(dir_relpath) - def _already_cached(self, path): - current_md5 = self.state.update(path) + def already_cached(self, path_info): + assert path_info['scheme'] in ['', 'local'] + + current_md5 = self.state.update(path_info['path']) if not current_md5: return False @@ -369,24 +345,7 @@ class RemoteLOCAL(RemoteBase): delta = working_dir_files - cached_files for file in delta: - if force or self._already_cached(file): - remove(file) - else: - self._safe_remove(file) - - def _safe_remove(self, file): - msg = ( - 'File "{}" is going to be removed. ' - 'Are you sure you want to proceed?' - .format(file) - ) - - if not prompt.confirm(msg): - raise DvcException('Unable to remove {} without a confirmation' - " from the user. Use '-f' to force." - .format(file)) - - remove(file) + self.safe_remove({'scheme': 'local', 'path': file}, force=force) def _move(self, inp, outp): # moving in two stages to make last the move atomic in diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 3fbacb4ca..3ca2ac5e9 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -115,7 +115,7 @@ class RemoteS3(RemoteBase): return checksum_info != self.save_info(path_info) - def _copy(self, from_info, to_info, s3=None): + def copy(self, from_info, to_info, s3=None): s3 = s3 if s3 else self.s3 source = {'Bucket': from_info['bucket'], @@ -130,46 +130,10 @@ class RemoteS3(RemoteBase): path = self.checksum_to_path(etag) to_info = {'scheme': 's3', 'bucket': self.bucket, 'path': path} - self._copy(path_info, to_info) + self.copy(path_info, to_info) return {self.PARAM_CHECKSUM: etag} - @staticmethod - def to_string(path_info): - return "s3://{}/{}".format(path_info['bucket'], path_info['path']) - - def checkout(self, path_info, checksum_info): - if path_info['scheme'] != 's3': - raise NotImplementedError - - etag = checksum_info.get(self.PARAM_CHECKSUM, None) - if not etag: - return - - if not self.changed(path_info, checksum_info): - msg = "Data '{}' didn't change." - logger.info(msg.format(self.to_string(path_info))) - return - - if self.changed_cache(etag): - msg = "Cache '{}' not found. File '{}' won't be created." - logger.warn(msg.format(etag, self.to_string(path_info))) - return - - if self.exists(path_info): - msg = "Data '{}' exists. Removing before checkout." - logger.warn(msg.format(self.to_string(path_info))) - self.remove(path_info) - return - - msg = "Checking out '{}' with cache '{}'." - logger.info(msg.format(self.to_string(path_info), etag)) - - path = self.checksum_to_path(etag) - from_info = {'scheme': 's3', 'bucket': self.bucket, 'path': path} - - self._copy(from_info, path_info) - def remove(self, path_info): if path_info['scheme'] != 's3': raise NotImplementedError @@ -262,7 +226,7 @@ class RemoteS3(RemoteBase): raise NotImplementedError if to_info['scheme'] == 's3': - self._copy(from_info, to_info, s3=s3) + self.copy(from_info, to_info, s3=s3) continue if to_info['scheme'] != 'local': diff --git a/dvc/remote/ssh.py b/dvc/remote/ssh.py index b7da48978..8d576cad1 100644 --- a/dvc/remote/ssh.py +++ b/dvc/remote/ssh.py @@ -188,7 +188,7 @@ class RemoteSSH(RemoteBase): return md5 - def cp(self, from_info, to_info, ssh=None): + def copy(self, from_info, to_info, ssh=None): if from_info['scheme'] != 'ssh' or to_info['scheme'] != 'ssh': raise NotImplementedError @@ -212,13 +212,6 @@ class RemoteSSH(RemoteBase): return {self.PARAM_CHECKSUM: self.md5(path_info)} - @staticmethod - def to_string(path_info): - return "{}://{}@{}:{}".format(path_info['scheme'], - path_info['user'], - path_info['host'], - path_info['path']) - def changed(self, path_info, checksum_info): if not self.exists(path_info): return True @@ -240,42 +233,10 @@ class RemoteSSH(RemoteBase): dest = path_info.copy() dest['path'] = self.checksum_to_path(md5) - self.cp(path_info, dest) + self.copy(path_info, dest) return {self.PARAM_CHECKSUM: md5} - def checkout(self, path_info, checksum_info): - if path_info['scheme'] != 'ssh': - raise NotImplementedError - - md5 = checksum_info.get(self.PARAM_CHECKSUM, None) - if not md5: - return - - if not self.changed(path_info, checksum_info): - msg = "Data '{}' didn't change." - logger.info(msg.format(self.to_string(path_info))) - return - - if self.changed_cache(md5): - msg = "Cache '{}' not found. File '{}' won't be created." - logger.warn(msg.format(md5, self.to_string(path_info))) - return - - if self.exists(path_info): - msg = "Data '{}' exists. Removing before checkout." - logger.warn(msg.format(self.to_string(path_info))) - self.remove(path_info) - return - - msg = "Checking out '{}' with cache '{}'." - logger.info(msg.format(self.to_string(path_info), md5)) - - src = path_info.copy() - src['path'] = self.checksum_to_path(md5) - - self.cp(src, path_info) - def remove(self, path_info): if path_info['scheme'] != 'ssh': raise NotImplementedError @@ -309,7 +270,7 @@ class RemoteSSH(RemoteBase): assert from_info['host'] == to_info['host'] assert from_info['port'] == to_info['port'] assert from_info['user'] == to_info['user'] - self.cp(from_info, to_info, ssh=ssh) + self.copy(from_info, to_info, ssh=ssh) continue if to_info['scheme'] != 'local':