dvc: cloud RFC add support for directories

Fixes #367

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

dvc/cloud/aws.py

@@ -103,6 +103,10 @@ class DataCloudAWS(DataCloudBase):
         Verify local and remote checksums. Used 'dvc-md5' metadata if supported
         or falls back to etag.
         """
+
+        # Calling get_key() once more to avoid empty metadata.
+        # See http://blog.bidiuk.com/2014/02/get-amazon-s3-metadata-in-python-using-boto/
+        key = key.bucket.get_key(key.name)
         md5_cloud = key.metadata.get('dvc-md5', None)
         md5_local = file_md5(fname)[0]
 
@@ -128,23 +132,21 @@ class DataCloudAWS(DataCloudBase):
         """
         return fname + '.download'
 
-    def _import(self, bucket_name, key_name, fname):
-
-        bucket = self._get_bucket_aws(bucket_name)
+    def _pull_key(self, key, fname):
+        Logger.debug("Pulling key '{}' from bucket '{}' to file '{}'".format(key.name,
+                                                                             key.bucket.name,
+                                                                             fname))
+        self._makedirs(fname)
 
         tmp_file = self.tmp_file(fname)
         name = os.path.basename(fname)
-        key = bucket.get_key(key_name)
-        if not key:
-            Logger.error('File "{}" does not exist in the cloud'.format(key_name))
-            return None
 
         if self._cmp_checksum(key, fname):
-            Logger.debug('File "{}" matches with "{}".'.format(fname, key_name))
+            Logger.debug('File "{}" matches with "{}".'.format(fname, key.name))
             return fname
 
-        Logger.debug('Downloading cache file from S3 "{}/{}" to "{}"'.format(bucket.name,
-                                                                             key_name,
+        Logger.debug('Downloading cache file from S3 "{}/{}" to "{}"'.format(key.bucket.name,
+                                                                             key.name,
                                                                              fname))
 
         res_h = ResumableDownloadHandler(tracker_file_name=self._download_tracker(tmp_file),
@@ -152,7 +154,7 @@ class DataCloudAWS(DataCloudBase):
         try:
             key.get_contents_to_filename(tmp_file, cb=create_cb(name), res_download_handler=res_h)
         except Exception as exc:
-            Logger.error('Failed to download "{}": {}'.format(key_name, exc))
+            Logger.error('Failed to download "{}": {}'.format(key.name, exc))
             return None
 
         os.rename(tmp_file, fname)
@@ -162,6 +164,24 @@ class DataCloudAWS(DataCloudBase):
 
         return fname
 
+    def _get_key(self, path):
+        key_name = self.cache_file_key(path)
+        bucket = self._get_bucket_aws(self.storage_bucket)
+        return bucket.get_key(key_name)
+
+    def _get_keys(self, path):
+        key_name = self.cache_file_key(path)
+        bucket = self._get_bucket_aws(self.storage_bucket)
+        keys = bucket.list(prefix=key_name)
+        if not keys:
+            return None
+        return list(keys)
+
+    def _new_key(self, path):
+        key_name = self.cache_file_key(path)
+        bucket = self._get_bucket_aws(self.storage_bucket)
+        return bucket.new_key(key_name)
+
     def _write_upload_tracker(self, fname, mp_id):
         """
         Write multipart id to upload tracker.
@@ -265,23 +285,8 @@ class DataCloudAWS(DataCloudBase):
         multipart.complete_upload()
         self._unlink_upload_tracker(fname)
 
-    def push(self, path):
+    def _push_key(self, key, path):
         """ push, aws version """
-
-        aws_key = self.cache_file_key(path)
-        bucket = self._get_bucket_aws(self.storage_bucket)
-        key = bucket.get_key(aws_key)
-        if key:
-            Logger.debug('File already uploaded to the cloud. Checksum validation...')
-
-            if self._cmp_checksum(key, path):
-                Logger.debug('File checksum matches. No uploading is needed.')
-                return path
-
-            Logger.debug('Checksum miss-match. Re-uploading is required.')
-
-        key = bucket.new_key(aws_key)
-
         try:
             self._push_multipart(key, path)
         except Exception as exc:
@@ -291,36 +296,3 @@ class DataCloudAWS(DataCloudBase):
         progress.finish_target(os.path.basename(path))
 
         return path
-
-    def _status(self, path):
-        aws_key = self.cache_file_key(path)
-        bucket = self._get_bucket_aws(self.storage_bucket)
-        key = bucket.get_key(aws_key)
-
-        remote_exists = key is not None
-        local_exists = os.path.exists(path)
-        diff = None
-        if remote_exists and local_exists:
-            diff = self._cmp_checksum(key, path)
-
-        return (local_exists, remote_exists, diff)
-
-    def remove(self, path):
-        aws_file_name = self.cache_file_key(path)
-
-        Logger.debug(u'[Cmd-Remove] Remove from cloud {}.'.format(aws_file_name))
-
-        if not self._aws_creds.access_key_id or not self._aws_creds.secret_access_key:
-            Logger.debug('[Cmd-Remove] Unable to check cache file in the cloud')
-            return
-        conn = S3Connection(self._aws_creds.access_key_id, self._aws_creds.secret_access_key)
-        bucket_name = self.storage_bucket
-        bucket = conn.lookup(bucket_name)
-        if bucket:
-            key = bucket.get_key(aws_file_name)
-            if not key:
-                Logger.warn('[Cmd-Remove] S3 remove warning: '
-                            'file "{}" does not exist in S3'.format(aws_file_name))
-            else:
-                key.delete()
-                Logger.info('[Cmd-Remove] File "{}" was removed from S3'.format(aws_file_name))

dvc/cloud/base.py

@@ -1,8 +1,12 @@
 import os
+import yaml
+import tempfile
+from checksumdir import dirhash
 
 from dvc.logger import Logger
 from dvc.exceptions import DvcException
 from dvc.config import ConfigError
+from dvc.stage import Output
 
 
 STATUS_UNKNOWN = 0
@@ -71,7 +75,9 @@ class DataCloudBase(object):
 
     def cache_file_key(self, fname):
         """ Key of a file within the bucket """
-        return '{}/{}'.format(self.storage_prefix, os.path.basename(fname)).strip('/')
+        relpath = os.path.relpath(fname, self._cloud_settings.cache_dir)
+        relpath.replace('\\', '/')
+        return '{}/{}'.format(self.storage_prefix, relpath).strip('/')
 
     @staticmethod
     def tmp_file(fname):
@@ -84,35 +90,141 @@ class DataCloudBase(object):
         """
         pass
 
-    def _import(self, bucket, fin, fout):
-        """
-        Cloud-specific method for importing data file.
-        """
+    def _push_key(self, key, path):
         pass
 
     def push(self, path):
-        """ Cloud-specific method for pushing data """
+        """ Generic method for pushing data """
+        if os.path.isfile(path):
+            return self._push(path)
+
+        res = []
+        for root, dirs, files in os.walk(path):
+            for fname in files:
+                p = os.path.join(root, fname)
+                res.append(self._push(p))
+
+                with open(p, 'r') as fd:
+                    d = yaml.safe_load(fd)
+                md5 = d[Output.PARAM_MD5]
+                cache = os.path.join(self._cloud_settings.cache_dir, md5)
+                res.append(self._push(cache))
+        return res
+
+    def _push(self, path):
+        key = self._get_key(path)
+        if key:
+            Logger.debug("File '{}' already uploaded to the cloud. Validating checksum...".format(path))
+            if self._cmp_checksum(key, path):
+                Logger.debug('File checksum matches. No uploading is needed.')
+                return []
+            Logger.debug('Checksum mismatch. Reuploading is required.')
+
+        key = self._new_key(path)
+        return [self._push_key(key, path)]
+
+    def _makedirs(self, fname):
+        dname = os.path.dirname(fname)
+        if not os.path.exists(dname):
+            os.makedirs(dname)
+
+    def _pull_key(self, key, path):
+        """ Cloud-specific method of pulling keys """
+        pass
+
+    def _get_key(self, path):
+        """ Cloud-specific method of getting keys """
+        pass
+
+    def _get_keys(self, path):
+        """ Cloud-specific method of getting keys """
         pass
 
     def pull(self, path):
         """ Generic method for pulling data from the cloud """
-        key_name = self.cache_file_key(path)
-        return self._import(self.storage_bucket, key_name, path)
+        key = self._get_key(path)
+        if key:
+            return [self._pull_key(key, path)]
+
+        keys = self._get_keys(path)
+        if not keys:
+            Logger.error("File '{}' does not exist in the cloud".format(path))
+            return []
+
+        res = []
+        for k in keys:
+            fname = os.path.join(path, os.path.relpath(k.name, self.cache_file_key(path)))
+            res.append(self._pull_key(k, fname))
+            with open(fname, 'r') as fd:
+                d = yaml.safe_load(fd)
+            md5 = d[Output.PARAM_MD5]
+
+            cache = os.path.join(self._cloud_settings.cache_dir, md5)
+            cache_key = self._get_key(cache)
+            if not cache_key:
+                Logger.error("File '{}' does not exist in the cloud".format(path))
+                continue
+
+            res.append(self._pull_key(cache_key, cache))
+
+        return res
 
     def remove(self, path):
         """
-        Cloud-specific method for removing data item from the cloud.
+        Generic method for removing data item from the cloud.
         """
-        pass
+        key = self._get_key(path)
+        if key:
+            key.delete()
+            return [k.name]
 
-    def _status(self, path):
-        """
-        Cloud-specific method for checking data item status.
-        """
-        pass
+        keys = self._get_keys(path)
+        if not keys:
+            Logger.error("File '{}' does not exist in the cloud".format(path))
+            return []
+
+        res = []
+        for k in keys:
+            k.delete()
+            res.append(k.name)
+        return res
+
+    def _status(self, key, path):
+        remote_exists = key != None
+        local_exists = os.path.exists(path)
+
+        diff = None
+        if remote_exists and local_exists:
+            diff = self._cmp_checksum(key, path)
+
+        return STATUS_MAP.get((local_exists, remote_exists, diff), STATUS_UNKNOWN)
 
     def status(self, path):
         """
         Generic method for checking data item status.
         """
-        return STATUS_MAP.get(self._status(path), STATUS_UNKNOWN)
+        key = self._get_key(path)
+        if key:
+            return self._status(key, path)
+
+        keys = self._get_keys(path)
+        if not keys or len(keys) == 0:
+            return STATUS_NEW
+
+        res = []
+        for k in keys:
+            fname = os.path.join(path, os.path.relpath(k.name, self.cache_file_key(path)))
+            res.append(self._status(k, fname))
+
+            tmp = tempfile.NamedTemporaryFile()
+            self._pull_key(k, tmp.name)
+            with open(tmp.name, 'r') as fd:
+                d = yaml.safe_load(fd)
+            tmp.close()
+            md5 = d[Output.PARAM_MD5]
+
+            cache = os.path.join(self._cloud_settings.cache_dir, md5)
+            cache_key = self._get_key(cache)
+            res.append(self._status(cache_key, cache))
+
+        return max(res)

dvc/cloud/data_cloud.py

@@ -7,7 +7,6 @@ from dvc.utils import map_progress
 from dvc.cloud.aws import DataCloudAWS
 from dvc.cloud.gcp import DataCloudGCP
 from dvc.cloud.local import DataCloudLOCAL
-from dvc.cloud.http import DataCloudHTTP
 
 
 class DataCloud(object):
@@ -16,20 +15,16 @@ class DataCloud(object):
     CLOUD_MAP = {
         'AWS'   : DataCloudAWS,
         'GCP'   : DataCloudGCP,
-        'HTTP'  : DataCloudHTTP,
         'LOCAL' : DataCloudLOCAL,
     }
 
     SCHEME_MAP = {
         's3'    : 'AWS',
-        'http'  : 'HTTP',
-        'https' : 'HTTP',
-        'ftp'   : 'HTTP',
         'gs'    : 'GCP',
         ''      : 'LOCAL',
     }
 
-    def __init__(self, config):
+    def __init__(self, cache_dir, config):
         self._config = config
 
         cloud_type = self._config['Global'].get('Cloud', '').strip().upper()
@@ -39,7 +34,8 @@ class DataCloud(object):
         if cloud_type not in self._config.keys():
             raise ConfigError('Can\'t find cloud section \'[%s]\' in config' % cloud_type)
 
-        cloud_settings = self.get_cloud_settings(self._config,
+        cloud_settings = self.get_cloud_settings(cache_dir,
+                                                 self._config,
                                                  cloud_type)
 
         self.typ = cloud_type
@@ -48,7 +44,7 @@ class DataCloud(object):
         self.sanity_check()
 
     @staticmethod
-    def get_cloud_settings(config, cloud_type):
+    def get_cloud_settings(cache_dir, config, cloud_type):
         """
         Obtain cloud settings from config.
         """
@@ -57,7 +53,7 @@ class DataCloud(object):
         else:
             cloud_config = config[cloud_type]
         global_storage_path = config['Global'].get('StoragePath', None)
-        cloud_settings = CloudSettings(global_storage_path, cloud_config)
+        cloud_settings = CloudSettings(cache_dir, global_storage_path, cloud_config)
         return cloud_settings
 
     def sanity_check(self):
@@ -89,12 +85,6 @@ class DataCloud(object):
         """
         return map_progress(func, targets, jobs)
 
-    def sync(self, targets, jobs=1):
-        """
-        Sync data items in a cloud-agnostic way.
-        """
-        return self._map_targets(self._cloud.sync, targets, jobs)
-
     def push(self, targets, jobs=1):
         """
         Push data items in a cloud-agnostic way.

dvc/cloud/gcp.py

@@ -47,84 +47,63 @@ class DataCloudGCP(DataCloudBase):
 
         return False
 
-    def _import(self, bucket_name, key, fname):
+    def _pull_key(self, key, path):
+        self._makedirs(path)
 
-        bucket = self._get_bucket_gc(bucket_name)
+        name = os.path.basename(path)
+        tmp_file = self.tmp_file(path)
 
-        name = os.path.basename(fname)
-        tmp_file = self.tmp_file(fname)
+        if self._cmp_checksum(key, path):
+            Logger.debug('File "{}" matches with "{}".'.format(path, key.name))
+            return path
 
-        blob = bucket.get_blob(key)
-        if not blob:
-            Logger.error('File "{}" does not exist in the cloud'.format(key))
-            return None
-
-        if self._cmp_checksum(blob, fname):
-            Logger.debug('File "{}" matches with "{}".'.format(fname, key))
-            return fname
-
-        Logger.debug('Downloading cache file from gc "{}/{}"'.format(bucket.name, key))
+        Logger.debug('Downloading cache file from gc "{}/{}"'.format(key.bucket.name, key.name))
 
         # percent_cb is not available for download_to_filename, so
         # lets at least update progress at keypoints(start, finish)
         progress.update_target(name, 0, None)
 
         try:
-            blob.download_to_filename(tmp_file)
+            key.download_to_filename(tmp_file)
         except Exception as exc:
-            Logger.error('Failed to download "{}": {}'.format(key, exc))
+            Logger.error('Failed to download "{}": {}'.format(key.name, exc))
             return None
 
-        os.rename(tmp_file, fname)
+        os.rename(tmp_file, path)
 
         progress.finish_target(name)
 
         Logger.debug('Downloading completed')
 
-        return fname
-
-    def push(self, path):
-        """ push, gcp version """
+        return path
 
+    def _get_key(self, path):
+        key_name = self.cache_file_key(path)
         bucket = self._get_bucket_gc(self.storage_bucket)
-        blob_name = self.cache_file_key(path)
+        return bucket.get_blob(key_name)
+
+    def _get_keys(self, path):
+        key_name = self.cache_file_key(path)
+        bucket = self._get_bucket_gc(self.storage_bucket)
+        keys = bucket.list_blobs(prefix=key_name)
+        if not keys:
+            return None
+        return list(keys)
+
+    def _new_key(self, path):
+        key_name = self.cache_file_key(path)
+        bucket = self._get_bucket_gc(self.storage_bucket)
+        return bucket.blob(key_name)
+
+    def _push_key(self, key, path):
+        """ push, gcp version """
         name = os.path.basename(path)
 
-        blob = bucket.get_blob(blob_name)
-        if blob is not None and blob.exists():
-            if self._cmp_checksum(blob, path):
-                Logger.debug('checksum %s matches.  Skipping upload' % path)
-                return path
-            Logger.debug('checksum %s mismatch.  re-uploading' % path)
-
-        # same as in _import
         progress.update_target(name, 0, None)
 
-        blob = bucket.blob(blob_name)
-        blob.upload_from_filename(path)
+        key.upload_from_filename(path)
 
         progress.finish_target(name)
         Logger.debug('uploading %s completed' % path)
 
         return path
-
-    def _status(self, path):
-        """ status, gcp version """
-
-        bucket = self._get_bucket_gc(self.storage_bucket)
-        blob_name = self.cache_file_key(path)
-        blob = bucket.get_blob(blob_name)
-
-        remote_exists = blob is not None and blob.exists()
-        local_exists = os.path.exists(path)
-        diff = None
-        if remote_exists and local_exists:
-            diff = self._cmp_checksum(blob, path)
-
-        return (local_exists, remote_exists, diff)
-
-    def remove(self, item):
-        bucket = self._get_bucket_gc(self.storage_bucket)
-        blob_name = self.cache_file_key(path)
-        blob = bucket.blob(blob_name)
-        blob.delete()

dvc/cloud/http.py

@@ -1,126 +0,0 @@
-import os
-import requests
-
-from dvc.logger import Logger
-from dvc.progress import progress
-from dvc.utils import copyfile
-from dvc.cloud.base import DataCloudError, DataCloudBase
-from dvc.utils import file_md5
-
-
-class DataCloudHTTP(DataCloudBase):
-    """
-    Driver for http cloud.
-    """
-    def push(self, path):
-        raise Exception('Not implemented yet')
-
-    def pull(self, path):
-        raise Exception('Not implemented yet')
-
-    def remove(self, path):
-        raise Exception('Not implemented yet')
-
-    def status(self, path):
-        raise Exception('Not implemented yet')
-
-    @staticmethod
-    def _downloaded_size(fname):
-        """
-        Check how much is already downloaded.
-        """
-        if os.path.exists(fname):
-            downloaded = os.path.getsize(fname)
-            header = {'Range': 'bytes=%d-' % downloaded}
-
-            Logger.debug('found existing {} file, resuming download'.format(fname))
-
-            return (downloaded, header)
-
-        return (0, None)
-
-    @staticmethod
-    def _get_header(req, name):
-        """
-        Get header value from request.
-        """
-        val = req.headers.get(name)
-        if val == None:
-            Logger.debug('\'{}\' not supported by the server'.format(name))
-
-        return val
-
-    def _verify_downloaded_size(self, req, downloaded_size):
-        """
-        Check that server supports resuming downloads.
-        """
-        content_range = self._get_header(req, 'content-range')
-        if downloaded_size and content_range == None:
-            Logger.debug('Can\'t resume download')
-            return 0
-
-        return downloaded_size
-
-    def _download(self, req, fname, downloaded):
-        """
-        Download file with progress bar.
-        """
-        mode = 'ab' if downloaded else 'wb'
-        name = os.path.basename(req.url)
-        total_length = self._get_header(req, 'content-length')
-        chunk_size = 1024 * 100
-
-        progress.update_target(name, downloaded, total_length)
-
-        with open(fname, mode) as fobj:
-            for chunk in req.iter_content(chunk_size=chunk_size):
-                if not chunk:  # filter out keep-alive new chunks
-                    continue
-
-                fobj.write(chunk)
-                downloaded += len(chunk)
-                progress.update_target(name, downloaded, total_length)
-
-
-        progress.finish_target(name)
-
-    def _verify_md5(self, req, fname):
-        """
-        Verify md5 of a downloaded file if server supports 'content-md5' header.
-        """
-        md5 = file_md5(fname)[0]
-        content_md5 = self._get_header(req, 'content-md5')
-
-        if content_md5 == None:
-            return True
-
-        if md5 != content_md5:
-            Logger.error('Checksum mismatch')
-            return False
-
-        Logger.debug('Checksum matches')
-        return True
-
-    def import_data(self, url, path):
-        """
-        Download single file from url.
-        """
-
-        tmp_file = self.tmp_file(path)
-
-        downloaded, header = self._downloaded_size(tmp_file)
-        req = requests.get(url, stream=True, headers=header)
-        downloaded = self._verify_downloaded_size(req, downloaded)
-
-        try:
-            self._download(req, tmp_file, downloaded)
-        except Exception as exc:
-            Logger.error('Failed to download "{}": {}'.format(url, exc))
-            return None
-
-        if not self._verify_md5(req, tmp_file):
-            return None
-
-        os.rename(tmp_file, path)
-
-        return path

dvc/cloud/instance_manager.py

@@ -1,4 +1,5 @@
 class CloudSettings(object):
-    def __init__(self, global_storage_path, cloud_config):
+    def __init__(self, cache_dir, global_storage_path, cloud_config):
+        self.cache_dir = cache_dir
         self.cloud_config = cloud_config
         self.global_storage_path = global_storage_path

dvc/cloud/local.py

@@ -6,48 +6,72 @@ from dvc.utils import copyfile, file_md5
 from dvc.cloud.base import DataCloudError, DataCloudBase
 
 
+class LocalKey(object):
+    def __init__(self, bucket, name):
+        self.name = name
+        self.bucket = bucket
+
+    @property
+    def path(self):
+        return os.path.join(self.bucket, self.name)
+
+    def delete(self):
+        os.unlink(self.path)
+
+
 class DataCloudLOCAL(DataCloudBase):
     """
     Driver for local storage.
     """
-    def push(self, path):
-        Logger.debug('sync to cloud ' + path + " " + self.storage_path)
-        copyfile(path, self.storage_path)
+    def cache_file_key(self, path):
+        return os.path.relpath(os.path.abspath(path), self._cloud_settings.cache_dir)
+
+    def _cmp_checksum(self, key, path):
+        return filecmp.cmp(path, key.path)
+
+    def _get_key(self, path):
+        key_name = self.cache_file_key(path)
+        key = LocalKey(self.storage_path, key_name)
+        if os.path.exists(key.path) and os.path.isfile(key.path):
+            return key
+        return None
+
+    def _get_keys(self, path):
+        key_name = self.cache_file_key(path)
+        key = LocalKey(self.storage_path, key_name)
+
+        if not os.path.isdir(key.path):
+            return None
+
+        res = []
+        for root, dirs, files in os.walk(key.path):
+            for fname in files:
+                p = os.path.join(root, fname)
+                k_name = os.path.relpath(p, self.storage_path)
+                res.append(LocalKey(self.storage_path, k_name))
+        return res
+
+    def _new_key(self, path):
+        key_name = self.cache_file_key(path)
+        key = LocalKey(self.storage_path, key_name)
+        self._makedirs(key.path)
+        return key
+
+    def _push_key(self, key, path):
+        self._makedirs(key.path)
+        copyfile(path, key.path)
         return path
 
-    def _import(self, bucket, i, path):
-        inp = os.path.join(self.storage_path, i)
+    def _pull_key(self, key, path):
+        self._makedirs(path)
+
         tmp_file = self.tmp_file(path)
         try:
-            copyfile(inp, tmp_file)
+            copyfile(key.path, tmp_file)
         except Exception as exc:
-            Logger.error('Failed to copy "{}": {}'.format(i, exc))
+            Logger.error('Failed to copy "{}": {}'.format(key.path, exc))
             return None
 
         os.rename(tmp_file, path)
 
         return path
-
-    def pull(self, path):
-        Logger.debug('sync from cloud ' + path)
-        return self._import(None, path, path)
-
-    def remove(self, path):
-        Logger.debug('rm from cloud ' + path)
-        os.remove(path)
-
-    def import_data(self, path, out):
-        Logger.debug('import from cloud ' + path + " " + out)
-        return self._import(None, path, out)
-
-    def _status(self, path):
-        local = path
-        remote = '{}/{}'.format(self.storage_path, os.path.basename(local))
-
-        remote_exists = os.path.exists(remote)
-        local_exists = os.path.exists(local)
-        diff = None
-        if local_exists and remote_exists:
-            diff = filecmp.cmp(local, remote)
-
-        return (local_exists, remote_exists, diff)

dvc/command/data_sync.py

@@ -1,4 +1,4 @@
-import dvc.cloud.data_cloud as cloud
+import dvc.cloud.base as cloud
 from dvc.command.common.base import CmdBase
 
 class CmdDataPull(CmdBase):

dvc/project.py

@@ -1,4 +1,5 @@
 import os
+import stat
 import networkx as nx
 
 from dvc.logger import Logger
@@ -77,7 +78,7 @@ class Project(object):
         self.state = State(self.root_dir, self.dvc_dir)
         self.config = Config(self.dvc_dir)
         self.logger = Logger(self.config._config)
-        self.cloud = DataCloud(self.config._config)
+        self.cloud = DataCloud(self.cache.cache_dir, self.config._config)
 
     @staticmethod
     def init(root_dir=os.curdir):
@@ -208,12 +209,30 @@ class Project(object):
                     clist.append(out.cache)
         return clist
 
+    def _remove_cache_file(self, cache):
+        os.chmod(cache, stat.S_IWRITE)
+        os.unlink(cache)
+
+    def _remove_cache(self, cache):
+        if os.path.isfile(cache):
+            self._remove_cache_file(cache)
+            return
+
+        for root, dirs, files in os.walk(cache, topdown=False):
+            for dname in dirs:
+                path = os.path.join(root, dname)
+                os.rmdir(path)
+            for fname in files:
+                path = os.path.join(root, fname)
+                self._remove_cache_file(path)
+        os.rmdir(cache)
+
     def gc(self):
         clist = self._used_cache()
         for cache in self.cache.all():
             if cache in clist:
                 continue
-            os.unlink(cache)
+            self._remove_cache(cache)
             self.logger.info(u'\'{}\' was removed'.format(self.to_dvc_path(cache)))
 
     def push(self, jobs=1):
@@ -221,10 +240,7 @@ class Project(object):
 
     def pull(self, jobs=1):
         self.cloud.pull(self._used_cache(), jobs)
-        for stage in self.stages():
-            for out in stage.outs:
-                if out.use_cache:
-                    out.link()
+        self.checkout()
 
     def status(self, jobs=1):
         return self.cloud.status(self._used_cache(), jobs)

dvc/stage.py

@@ -3,6 +3,7 @@ import stat
 import yaml
 import itertools
 import subprocess
+from checksumdir import dirhash
 
 from dvc.system import System
 from dvc.utils import file_md5
@@ -45,9 +46,9 @@ class CmdOutputDoesNotExistError(CmdOutputError):
         super(CmdOutputDoesNotExistError, self).__init__(path, 'does not exist')
 
 
-class CmdOutputIsNotFileError(CmdOutputError):
+class CmdOutputIsNotFileOrDirError(CmdOutputError):
     def __init__(self, path):
-        super(CmdOutputIsNotFileError, self).__init__(path, 'not a file')
+        super(CmdOutputIsNotFileOrDirError, self).__init__(path, 'not a file or directory')
 
 
 class CmdOutputAlreadyTrackedError(CmdOutputError):
@@ -84,13 +85,19 @@ class Dependency(object):
         if state and state.mtime == self.mtime():
             md5 = state.md5
         else:
-            md5 = file_md5(self.path)[0]
+            md5 = self.compute_md5()
 
         return self.md5 != md5
 
     def changed(self):
         return self._changed_md5()
 
+    def compute_md5(self):
+        if os.path.isdir(self.path):
+            return dirhash(self.path, hashfunc='md5')
+        else:
+            return file_md5(self.path)[0]
+
     def mtime(self):
         return os.path.getmtime(self.path)
 
@@ -100,8 +107,9 @@ class Dependency(object):
     def update(self):
         if not os.path.exists(self.path):
             raise CmdOutputDoesNotExistError(self.rel_path)
-        if not os.path.isfile(self.path):
-            raise CmdOutputIsNotFileError(self.path)
+
+        if not os.path.isfile(self.path) and not os.path.isdir(self.path):
+            raise CmdOutputIsNotFileOrDirError(self.rel_path)
 
         state = self.project.state.get(self.path)
         if state and state.mtime == self.mtime() and state.inode == self.inode():
@@ -110,7 +118,7 @@ class Dependency(object):
             self.project.logger.debug(msg.format(self.path, md5))
             self.md5 = md5
         else:
-            self.md5 = file_md5(self.path)[0]
+            self.md5 = self.compute_md5()
             self.project.state.update(self.path, self.md5, self.mtime(), self.inode())
 
     def dumpd(self, cwd):
@@ -189,16 +197,32 @@ class Output(Dependency):
 
         return True
 
-    @staticmethod
-    def hardlink(src, link):
+    def hardlink(self, src, link):
+        self.project.logger.debug("creating hardlink {} -> {}".format(src, link))
         System.hardlink(src, link)
         os.chmod(src, stat.S_IREAD)
 
+    def dir_cache(self):
+        res = {}
+        for root, dirs, files in os.walk(self.cache):
+            for fname in files:
+                path = os.path.join(root, fname)
+                relpath = os.path.relpath(path, self.cache)
+                with open(path, 'r') as fd:
+                    d = yaml.safe_load(fd)
+                md5 = d[Output.PARAM_MD5]
+                res[relpath] = self.project.cache.get(md5)
+        return res
+
     def checkout(self):
         if not self.use_cache:
             return
 
+        self.project.logger.debug("Checking out {} with cache {}".format(self.path, self.cache))
+
         if not self.changed():
+            msg = "Data {} with cache {} didn't change"
+            self.project.logger.debug(msg.format(self.path, self.cache))
             return
 
         if not os.path.exists(self.cache):
@@ -207,45 +231,92 @@ class Output(Dependency):
             return
 
         if os.path.exists(self.path):
+            msg = "Data {} exists. Removing before checkout"
+            self.project.logger.debug(msg.format(self.path))
             self.remove()
 
-        self.hardlink(self.cache, self.path)
+        if os.path.isfile(self.cache):
+            self.hardlink(self.cache, self.path)
+            return
+
+        for relpath, cache in self.dir_cache().items():
+            path = os.path.join(self.path, relpath)
+            os.makedirs(os.path.dirname(path))
+            self.hardlink(cache, path)
 
     def save(self):
         if not self.use_cache:
             return
 
+        self.project.logger.debug("Saving {} to {}".format(self.path, self.cache))
+
         if not os.path.exists(self.path):
             raise CmdOutputDoesNotExistError(self.rel_path)
 
-        if not os.path.isfile(self.path):
-            raise CmdOutputIsNotFileError(self.rel_path)
+        if not os.path.isfile(self.path) and not os.path.isdir(self.path):
+            raise CmdOutputIsNotFileOrDirError(self.rel_path)
 
         if self.project.scm.is_tracked(self.path):
-            raise CmdOutputAlreadyTrackedError(self.path)
+            raise CmdOutputAlreadyTrackedError(self.rel_path)
 
         if not self.changed():
+            msg = "Data {} with cache {} didn't change"
+            self.project.logger.debug(msg.format(self.path, self.cache))
             return
 
         if os.path.exists(self.cache):
             # This means that we already have cache for this data.
             # We remove data and link it to existing cache to save
             # some space.
-            self.remove()
-            self.hardlink(self.cache, self.path)
+            msg = "Cache {} already exists, performing checkout for {}"
+            self.project.logger.debug(msg.format(self.cache, self.path))
+            self.checkout()
             return
 
-        self.hardlink(self.path, self.cache)
+        if os.path.isfile(self.path):
+            self.hardlink(self.path, self.cache)
+            return
+
+        for root, dirs, files in os.walk(self.path):
+            for fname in files:
+                path = os.path.join(root, fname)
+                relpath = os.path.relpath(path, self.path)
+                md5 = file_md5(path)[0]
+                cache = self.project.cache.get(md5)
+                cache_info = os.path.join(self.cache, relpath)
+
+                self.hardlink(path, cache)
+
+                os.makedirs(os.path.dirname(cache_info))
+                with open(cache_info, 'w') as fd:
+                    yaml.safe_dump({self.PARAM_MD5: md5}, fd, default_flow_style=False)
+
+    def _remove(self, path, cache):
+        self.project.logger.debug("Removing '{}'".format(path))
+        os.chmod(path, stat.S_IWUSR)
+        os.unlink(path)
+        if cache != None and os.path.exists(cache):
+            os.chmod(cache, stat.S_IREAD)
 
     def remove(self):
         if not os.path.exists(self.path):
             return
 
-        self.project.logger.debug("Removing '{}'".format(self.path))
-        os.chmod(self.path, stat.S_IWUSR)
-        os.unlink(self.path)
-        if os.path.exists(self.cache):
-            os.chmod(self.cache, stat.S_IREAD)
+        if os.path.isfile(self.path):
+            self._remove(self.path, self.cache)
+            return
+
+        caches = self.dir_cache()
+        for root, dirs, files in os.walk(self.path, topdown=False):
+            for d in dirs:
+                path = os.path.join(root, d)
+                os.rmdir(path)
+            for f in files:
+                path = os.path.join(root, f)
+                relpath = os.path.relpath(path, self.path)
+                cache = caches.get(relpath, None)
+                self._remove(path, cache)
+        os.rmdir(self.path)
 
 
 class StageCmdFailedError(DvcException):

dvc/utils.py

@@ -83,7 +83,8 @@ def map_progress(func, targets, n_threads):
     try:
         ret = pool.map(func, targets)
     except Exception as exc:
-        Logger.error('Unexpected exception while processing targets: {}'.format(exc), exc_info=True)
+#        Logger.error('Unexpected exception while processing targets: {}'.format(exc), exc_info=True)
+        raise
     finally:
         progress.finish()
 

requirements.txt

@@ -95,3 +95,4 @@ tinydb==3.7.0
 gitpython==2.1.8
 ntfsutils==0.1.3
 pygraphviz==1.3.1
+checksumdir==1.1.4

setup.py

@@ -100,6 +100,7 @@ install_requires = [
     'gitpython==2.1.8',
     'ntfsutils==0.1.3',
     'pygraphviz==1.3.1',
+    'checksumdir==1.1.4',
 ]
 
 setup(

tests/basic_env.py

@@ -9,6 +9,12 @@ from dvc.project import Project
 
 
 class TestDir(TestCase):
+    DATA_DIR = 'data_dir'
+    DATA_SUB_DIR = os.path.join(DATA_DIR, 'data_sub_dir')
+    DATA = os.path.join(DATA_DIR, 'data')
+    DATA_SUB = os.path.join(DATA_SUB_DIR, 'data_sub')
+    DATA_CONTENTS = DATA
+    DATA_SUB_CONTENTS = DATA_SUB
     FOO = 'foo'
     FOO_CONTENTS = FOO
     BAR = 'bar'
@@ -34,6 +40,10 @@ class TestDir(TestCase):
         self.create(self.FOO, self.FOO_CONTENTS)
         self.create(self.BAR, self.BAR_CONTENTS)
         self.create(self.CODE, self.CODE_CONTENTS)
+        os.mkdir(self.DATA_DIR)
+        os.mkdir(self.DATA_SUB_DIR)
+        self.create(self.DATA, self.DATA_CONTENTS)
+        self.create(self.DATA_SUB, self.DATA_SUB_CONTENTS)
 
     def tearDown(self):
         self._popd()

tests/test_add.py

@@ -1,11 +1,12 @@
 import os
 import shutil
 import filecmp
+from checksumdir import dirhash
 
 from dvc.main import main
 from dvc.utils import file_md5
 from dvc.stage import Stage, CmdOutputNoCacheError, CmdOutputOutsideOfRepoError
-from dvc.stage import CmdOutputDoesNotExistError, CmdOutputIsNotFileError
+from dvc.stage import CmdOutputDoesNotExistError, CmdOutputIsNotFileOrDirError
 from dvc.stage import CmdOutputAlreadyTrackedError
 from dvc.project import StageNotFoundError
 from dvc.command.add import CmdAdd
@@ -43,8 +44,8 @@ class TestAddDirectory(TestDvc):
     def test(self):
         dname = 'directory'
         os.mkdir(dname)
-        with self.assertRaises(CmdOutputIsNotFileError) as cx:
-            self.dvc.add(dname)
+        self.create(os.path.join(dname, 'file'), 'file')
+        self.dvc.add(dname)
 
 
 class TestAddTrackedFile(TestDvc):

tests/test_config.py

@@ -46,7 +46,7 @@ class TestConfigTest(TestCase):
            )
         s = StringIO('\n'.join(c))
         conf = self._conf(s)
-        cloud = DataCloud(conf)
+        cloud = DataCloud(None, conf)
         self.assertEqual(cloud.typ, 'AWS')
         self.assertEqual(cloud._cloud.storage_bucket, 'globalsb')
         self.assertEqual(cloud._cloud.storage_prefix, 'global_storage_path')
@@ -65,7 +65,7 @@ class TestConfigTest(TestCase):
            )
         s = StringIO('\n'.join(c))
         conf = self._conf(s)
-        cloud = DataCloud(conf)
+        cloud = DataCloud(None, conf)
         self.assertEqual(cloud.typ, 'AWS')
         self.assertEqual(cloud._cloud.storage_bucket, 'awssb')
         self.assertEqual(cloud._cloud.storage_prefix, 'aws_storage_path')
@@ -104,7 +104,7 @@ class TestConfigTest(TestCase):
 
         # patcher.start()
         conf = self._conf(s)
-        cloud = DataCloud(conf)
+        cloud = DataCloud(None, conf)
         aws_creds = cloud._cloud._get_credentials()
         patcher.stop()
 
@@ -135,7 +135,7 @@ class TestConfigTest(TestCase):
 
         patcher.start()
         conf = self._conf(s)
-        cloud = DataCloud(conf)
+        cloud = DataCloud(None, conf)
         aws_creds = cloud._cloud._get_credentials()
         patcher.stop()
 

tests/test_data_cloud.py

@@ -1,8 +1,9 @@
 import os
+import stat
 import uuid
 
 from dvc.config import ConfigError
-from dvc.cloud.data_cloud import DataCloud, DataCloudAWS, DataCloudLOCAL, DataCloudHTTP, DataCloudGCP
+from dvc.cloud.data_cloud import DataCloud, DataCloudAWS, DataCloudLOCAL, DataCloudGCP
 from dvc.cloud.base import STATUS_UNKNOWN, STATUS_OK, STATUS_MODIFIED, STATUS_NEW, STATUS_DELETED
 from dvc.cloud.instance_manager import CloudSettings
 
@@ -15,12 +16,12 @@ class TestDataCloud(TestDvc):
             config = {'Global': {'Cloud': k},
                       k: {'StoragePath': 'a/b',
                           'ProjectName': 'name'}}
-            cloud = DataCloud(config)
+            cloud = DataCloud(self.dvc.cache.cache_dir, config)
             self.assertIsInstance(cloud._cloud, v)
 
         with self.assertRaises(ConfigError) as cx:
             config = {'Global': {'Cloud': 'not_supported_type'}}
-            DataCloud(config)
+            DataCloud(self.dvc.cache.cache_dir, config)
 
 
 class TestDataCloudBase(TestDvc):
@@ -30,45 +31,77 @@ class TestDataCloudBase(TestDvc):
     def _test_cloud(self):
         self._setup_cloud()
 
+        stage = self.dvc.add(self.FOO)
+        cache = stage.outs[0].cache
+
+        stage_dir = self.dvc.add(self.DATA_DIR)
+        cache_dir = stage_dir.outs[0].cache
+
         # Check status
-        status = self.cloud.status(self.FOO)
+        status = self.cloud.status(cache)
         self.assertEqual(status, STATUS_NEW)
 
-        # Push and check status
-        self.cloud.push(self.FOO)
-        self.assertTrue(os.path.exists(self.FOO))
-        self.assertTrue(os.path.isfile(self.FOO))
+        status_dir = self.cloud.status(cache_dir)
+        self.assertEqual(status_dir, STATUS_NEW)
 
-        status = self.cloud.status(self.FOO)
+        # Push and check status
+        self.cloud.push(cache)
+        self.assertTrue(os.path.exists(cache))
+        self.assertTrue(os.path.isfile(cache))
+
+        self.cloud.push(cache_dir)
+        self.assertTrue(os.path.exists(cache_dir))
+        self.assertTrue(os.path.isdir(cache_dir))
+
+        status = self.cloud.status(cache)
         self.assertEqual(status, STATUS_OK)
 
+        status_dir = self.cloud.status(cache_dir)
+        self.assertEqual(status_dir, STATUS_OK)
+
         # Modify and check status
-        with open(self.FOO, 'a') as fd:
+        os.chmod(cache, stat.S_IWRITE)
+        with open(cache, 'a') as fd:
             fd.write('addon')
-        status = self.cloud.status(self.FOO)
+        os.chmod(cache, 0o777)
+        status = self.cloud.status(cache)
         self.assertEqual(status, STATUS_MODIFIED)
 
         # Remove and check status
-        os.unlink(self.FOO)
-        status = self.cloud.status(self.FOO)
+        os.unlink(cache)
+        status = self.cloud.status(cache)
         self.assertEqual(status, STATUS_DELETED)
 
+        self.dvc._remove_cache(cache_dir)
+        status_dir = self.cloud.status(cache_dir)
+        self.assertEqual(status_dir, STATUS_DELETED)
+
         # Pull and check status
-        self.cloud.pull(self.FOO)
-        self.assertTrue(os.path.exists(self.FOO))
-        self.assertTrue(os.path.isfile(self.FOO))
-        with open(self.FOO, 'r') as fd:
+        self.cloud.pull(cache)
+        self.assertTrue(os.path.exists(cache))
+        self.assertTrue(os.path.isfile(cache))
+        with open(cache, 'r') as fd:
             self.assertEqual(fd.read(), self.FOO_CONTENTS)
 
-        status = self.cloud.status(self.FOO)
+        self.cloud.pull(cache_dir)
+        self.assertTrue(os.path.exists(cache_dir))
+        self.assertTrue(os.path.isdir(cache_dir))
+
+        status = self.cloud.status(cache)
         self.assertEqual(status, STATUS_OK)
 
+        status_dir = self.cloud.status(cache_dir)
+        self.assertEqual(status_dir, STATUS_OK)
+
 
 class TestDataCloudAWS(TestDataCloudBase):
     TEST_REPO_BUCKET = 'dvc-test'
     TEST_REPO_REGION = 'us-east-2'
 
     def _should_test(self):
+        if os.name == 'nt':
+            return False
+
         if os.getenv("TRAVIS_PULL_REQUEST") == "false" and \
            os.getenv("TRAVIS_SECURE_ENV_VARS") == "true":
             return True
@@ -83,7 +116,7 @@ class TestDataCloudAWS(TestDataCloudBase):
         # Setup cloud
         config = {'StoragePath': repo,
                   'Region': self.TEST_REPO_REGION}
-        cloud_settings = CloudSettings(None, config)
+        cloud_settings = CloudSettings(self.dvc.cache.cache_dir, None, config)
         self.cloud = DataCloudAWS(cloud_settings)
 
     def test(self):
@@ -96,6 +129,9 @@ class TestDataCloudGCP(TestDataCloudBase):
     TEST_REPO_GCP_PROJECT='dvc-project'
 
     def _should_test(self):
+        if os.name == 'nt':
+            return False
+
         if os.getenv("TRAVIS") == "true" or os.getenv("APPVEYOR") == "True" or os.getenv("DVC_TEST_GCP") != "true":
             return False
         return True
@@ -110,7 +146,7 @@ class TestDataCloudGCP(TestDataCloudBase):
         # Setup cloud
         config = {'StoragePath': self.TEST_REPO_GCP,
                   'Region': self.TEST_REPO_GCP_PROJECT}
-        cloud_settings = CloudSettings(None, config)
+        cloud_settings = CloudSettings(self.dvc.cache.cache_dir, None, config)
         self.cloud = DataCloudGCP(cloud_settings)
 
     def test(self):
@@ -119,26 +155,19 @@ class TestDataCloudGCP(TestDataCloudBase):
 
 
 class TestDataCloudLOCAL(TestDataCloudBase):
+    def _should_test(self):
+        if os.name == 'nt':
+            return False
+        return True
+
     def _setup_cloud(self):
         self.dname = 'cloud'
         os.mkdir(self.dname)
         config = {'StoragePath': self.dname}
-        cloud_settings = CloudSettings(None, config)
+        cloud_settings = CloudSettings(self.dvc.cache.cache_dir, None, config)
         self.cloud = DataCloudLOCAL(cloud_settings)
 
     def test(self):
-        self._test_cloud()
-        self.assertTrue(os.path.isdir(self.dname))
-        path = os.path.join(self.dname, self.FOO)
-        self.assertTrue(os.path.isfile(path))
-
-
-class TestDataCloudHTTP(TestDvc):
-    def test(self):
-        cloud = DataCloudHTTP(CloudSettings(None, None))
-        path = 'out'
-        url = 'https://github.com/dataversioncontrol/dvc/LICENSE'
-        ret = cloud.import_data(url, path)
-        self.assertEqual(ret, path)
-        self.assertTrue(os.path.exists(path))
-        self.assertTrue(os.path.isfile(path))
+        if self._should_test():
+            self._test_cloud()
+            self.assertTrue(os.path.isdir(self.dname))

tests/test_gc.py

@@ -5,7 +5,7 @@ import filecmp
 from dvc.main import main
 from dvc.utils import file_md5
 from dvc.stage import Stage, CmdOutputNoCacheError, CmdOutputOutsideOfRepoError
-from dvc.stage import CmdOutputDoesNotExistError, CmdOutputIsNotFileError
+from dvc.stage import CmdOutputDoesNotExistError
 from dvc.project import StageNotFoundError
 from dvc.command.gc import CmdGC