remote: RFC: unify checkout()

Signed-off-by: Ruslan Kuprieiev <[email protected]>

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)

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':

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':

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

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':

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':