Merge pull request #512 from efiop/master

Pull/push per dvc-file

dvc/cli.py

@@ -132,6 +132,12 @@ def parse_args(argv=None):
                         type=int,
                         default=cpu_count(),
                         help='Number of jobs to run simultaneously.')
+    parent_cache_parser.add_argument(
+                        'targets',
+                        nargs='*',
+                        default=None,
+                        help='DVC files.')
+
     # Cache
     cache_parser = subparsers.add_parser(
                         'cache',

dvc/cloud/aws.py

@@ -130,7 +130,7 @@ class DataCloudAWS(DataCloudBase):
         """
         return fname + '.download'
 
-    def _pull_key(self, key, fname):
+    def _pull_key(self, key, fname, no_progress_bar=False):
         Logger.debug("Pulling key '{}' from bucket '{}' to file '{}'".format(key.name,
                                                                              key.bucket.name,
                                                                              fname))
@@ -147,17 +147,24 @@ class DataCloudAWS(DataCloudBase):
                                                                              key.name,
                                                                              fname))
 
+        if no_progress_bar:
+            cb = None
+        else:
+            cb = create_cb(name)
+
         res_h = ResumableDownloadHandler(tracker_file_name=self._download_tracker(tmp_file),
                                          num_retries=10)
         try:
-            key.get_contents_to_filename(tmp_file, cb=create_cb(name), res_download_handler=res_h)
+            key.get_contents_to_filename(tmp_file, cb=cb, res_download_handler=res_h)
         except Exception as exc:
             Logger.error('Failed to download "{}": {}'.format(key.name, exc))
             return None
 
         os.rename(tmp_file, fname)
 
-        progress.finish_target(name)
+        if not no_progress_bar:
+            progress.finish_target(name)
+
         Logger.debug('Downloading completed')
 
         return fname

dvc/cloud/base.py

@@ -109,7 +109,7 @@ class DataCloudBase(object):
                 Logger.debug("File '{}' does not exist in the cloud".format(path))
                 return ret
             tmp = os.path.join(tempfile.mkdtemp(), os.path.basename(path))
-            self._pull_key(key, tmp)
+            self._pull_key(key, tmp, no_progress_bar=True)
             dir_path = tmp
 
         for relpath, md5 in Output.get_dir_cache(dir_path).items():
@@ -138,7 +138,7 @@ class DataCloudBase(object):
             if e.errno != os.errno.EEXIST:
                 raise
 
-    def _pull_key(self, key, path):
+    def _pull_key(self, key, path, no_progress_bar=False):
         """ Cloud-specific method of pulling keys """
         pass
 

dvc/cloud/gcp.py

@@ -45,7 +45,7 @@ class DataCloudGCP(DataCloudBase):
 
         return False
 
-    def _pull_key(self, key, path):
+    def _pull_key(self, key, path, no_progress_bar=False):
         self._makedirs(path)
 
         name = os.path.relpath(path, self._cloud_settings.cache.cache_dir)
@@ -57,9 +57,10 @@ class DataCloudGCP(DataCloudBase):
 
         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)
+        if not no_progress_bar:
+            # 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:
             key.download_to_filename(tmp_file)
@@ -69,7 +70,8 @@ class DataCloudGCP(DataCloudBase):
 
         os.rename(tmp_file, path)
 
-        progress.finish_target(name)
+        if not no_progress_bar:
+            progress.finish_target(name)
 
         Logger.debug('Downloading completed')
 

dvc/cloud/local.py

@@ -47,7 +47,7 @@ class DataCloudLOCAL(DataCloudBase):
         copyfile(path, key.path)
         return path
 
-    def _pull_key(self, key, path):
+    def _pull_key(self, key, path, no_progress_bar=False):
         self._makedirs(path)
 
         tmp_file = self.tmp_file(path)

dvc/command/data_sync.py

@@ -6,27 +6,41 @@ from dvc.command.common.base import CmdBase
 from dvc.exceptions import DvcException
 
 
-class CmdDataPull(CmdBase):
+class CmdDataBase(CmdBase):
+    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
+
+class CmdDataPull(CmdDataBase):
+    def do_run(self, target=None):
         try:
-            self.project.pull(self.args.jobs)
+            self.project.pull(target=target, jobs=self.args.jobs)
         except DvcException as exc:
             self.project.logger.error('Failed to pull data from the cloud', exc)
             return 1
         return 0
 
 
-class CmdDataPush(CmdBase):
-    def run(self):
+class CmdDataPush(CmdDataBase):
+    def do_run(self, target=None):
         try:
-            self.project.push(self.args.jobs)
+            self.project.push(target=target, jobs=self.args.jobs)
         except DvcException as exc:
             self.project.logger.error('Failed to push data to the cloud', exc)
             return 1
         return 0
 
 
-class CmdDataStatus(CmdBase):
+class CmdDataStatus(CmdDataBase):
     def _show(self, status):
         for s in status:
             target, ret = s
@@ -44,9 +58,9 @@ class CmdDataStatus(CmdBase):
 
             self.project.logger.info('\t{}\t{}'.format(prefix_map[ret], path))
 
-    def run(self):
+    def do_run(self, target=None):
         try:
-            status = self.project.status(self.args.jobs)
+            status = self.project.status(target=target, jobs=self.args.jobs)
             self._show(status)
         except DvcException as exc:
             self.project.logger.error('Failed to obtain data status', exc)

dvc/project.py

@@ -159,19 +159,24 @@ class Project(object):
         for stage in self.stages():
             stage.checkout()
 
-    def _used_cache(self):
-        clist = []
-        for stage in self.stages():
+    def _used_cache(self, target=None):
+        cache_set = set()
+
+        if target:
+            stages = [Stage.load(self, target)]
+        else:
+            stages = self.stages()
+
+        for stage in stages:
             for out in stage.outs:
                 if not out.use_cache:
                     continue
-                if out.cache not in clist:
-                    clist.append(out.cache)
+                cache_set |= set([out.cache])
                 if out.is_dir_cache(out.cache) and os.path.isfile(out.cache):
                     dir_cache = out.dir_cache()
-                    clist.extend(dir_cache.values())
-                    clist.append(out.cache)
-        return clist
+                    cache_set |= set(dir_cache.values())
+
+        return list(cache_set)
 
     def _remove_cache_file(self, cache):
         os.chmod(cache, stat.S_IWRITE)
@@ -199,16 +204,16 @@ class Project(object):
             self._remove_cache(cache)
             self.logger.info(u'\'{}\' was removed'.format(self.to_dvc_path(cache)))
 
-    def push(self, jobs=1):
-        return self.cloud.push(self._used_cache(), jobs)
+    def push(self, target=None, jobs=1):
+        return self.cloud.push(self._used_cache(target), jobs)
 
-    def pull(self, jobs=1):
-        ret = self.cloud.pull(self._used_cache(), jobs)
+    def pull(self, target=None, jobs=1):
+        ret = self.cloud.pull(self._used_cache(target), jobs)
         self.checkout()
         return ret
 
-    def status(self, jobs=1):
-        return self.cloud.status(self._used_cache(), jobs)
+    def status(self, target=None, jobs=1):
+        return self.cloud.status(self._used_cache(target), jobs)
 
     def graph(self):
         G = nx.DiGraph()