diff --git a/dvc/cli.py b/dvc/cli.py index 949a79bc5..7703e445a 100644 --- a/dvc/cli.py +++ b/dvc/cli.py @@ -355,6 +355,11 @@ def parse_args(argv=None): default=False, help='Do not prompt when removing ' 'working directory files.') + pull_parser.add_argument( + '-R', + '--recursive', + help='Pull cache for subdirectories of the specified ' + 'directory.') pull_parser.set_defaults(func=CmdDataPull) # Push @@ -387,6 +392,11 @@ def parse_args(argv=None): default=False, help='Push cache for all dependencies of the ' 'specified target.') + push_parser.add_argument( + '-R', + '--recursive', + help='Push cache from subdirectories of specified ' + 'directory.') push_parser.set_defaults(func=CmdDataPush) # Fetch @@ -419,6 +429,11 @@ def parse_args(argv=None): default=False, help='Fetch cache for all dependencies of the ' 'specified target.') + fetch_parser.add_argument( + '-R', + '--recursive', + help='Fetch cache for subdirectories of specified ' + 'directory.') fetch_parser.set_defaults(func=CmdDataFetch) # Status diff --git a/dvc/command/data_sync.py b/dvc/command/data_sync.py index 0fa44e162..31398d2d0 100644 --- a/dvc/command/data_sync.py +++ b/dvc/command/data_sync.py @@ -27,7 +27,8 @@ class CmdDataPull(CmdDataBase): all_branches=self.args.all_branches, all_tags=self.args.all_tags, with_deps=self.args.with_deps, - force=self.args.force) + force=self.args.force, + from_directory=self.args.recursive) except Exception: logger.error('failed to pull data from the cloud') return 1 @@ -43,7 +44,8 @@ class CmdDataPush(CmdDataBase): show_checksums=self.args.show_checksums, all_branches=self.args.all_branches, all_tags=self.args.all_tags, - with_deps=self.args.with_deps) + with_deps=self.args.with_deps, + from_directory=self.args.recursive) except Exception: logger.error('failed to push data to the cloud') return 1 @@ -59,7 +61,8 @@ class CmdDataFetch(CmdDataBase): show_checksums=self.args.show_checksums, all_branches=self.args.all_branches, all_tags=self.args.all_tags, - with_deps=self.args.with_deps) + with_deps=self.args.with_deps, + from_directory=self.args.recursive) except Exception: logger.error('failed to fetch data from the cloud') return 1 diff --git a/dvc/project.py b/dvc/project.py index 04fb30a69..a3398680a 100644 --- a/dvc/project.py +++ b/dvc/project.py @@ -554,8 +554,12 @@ class Project(object): used.append(out.path) self.state.remove_unused_links(used) - def checkout(self, target=None, with_deps=False, force=False): - all_stages = self.active_stages() + def checkout(self, + target=None, + with_deps=False, + force=False, + from_directory=None): + all_stages = self.active_stages(from_directory) stages = all_stages if target: @@ -680,7 +684,8 @@ class Project(object): all_tags=False, remote=None, force=False, - jobs=None): + jobs=None, + from_directory=None): cache = {} cache['local'] = [] cache['s3'] = [] @@ -695,9 +700,9 @@ class Project(object): stages = self._collect(target, with_deps=with_deps) elif active: - stages = self.active_stages() + stages = self.active_stages(from_directory) else: - stages = self.stages() + stages = self.stages(from_directory) for stage in stages: if active and not target and stage.locked: @@ -820,7 +825,8 @@ class Project(object): all_branches=False, show_checksums=False, with_deps=False, - all_tags=False): + all_tags=False, + from_directory=None): with self.state: used = self._used_cache(target, all_branches=all_branches, @@ -828,7 +834,8 @@ class Project(object): with_deps=with_deps, force=True, remote=remote, - jobs=jobs)['local'] + jobs=jobs, + from_directory=from_directory)['local'] self.cloud.push(used, jobs, remote=remote, @@ -841,7 +848,8 @@ class Project(object): all_branches=False, show_checksums=False, with_deps=False, - all_tags=False): + all_tags=False, + from_directory=None): with self.state: used = self._used_cache(target, all_branches=all_branches, @@ -849,7 +857,8 @@ class Project(object): with_deps=with_deps, force=True, remote=remote, - jobs=jobs)['local'] + jobs=jobs, + from_directory=from_directory)['local'] self.cloud.pull(used, jobs, remote=remote, @@ -863,15 +872,20 @@ class Project(object): show_checksums=False, with_deps=False, all_tags=False, - force=False): + force=False, + from_directory=None): self.fetch(target, jobs, remote=remote, all_branches=all_branches, all_tags=all_tags, show_checksums=show_checksums, - with_deps=with_deps) - self.checkout(target=target, with_deps=with_deps, force=force) + with_deps=with_deps, + from_directory=from_directory) + self.checkout(target=target, + with_deps=with_deps, + force=force, + from_directory=from_directory) def _local_status(self, target=None, with_deps=False): status = {} @@ -1139,13 +1153,13 @@ class Project(object): def metrics_remove(self, path): self._metrics_modify(path, delete=True) - def graph(self): + def graph(self, from_directory=None): import networkx as nx from dvc.exceptions import OutputDuplicationError G = nx.DiGraph() G_active = nx.DiGraph() - stages = self.stages() + stages = self.stages(from_directory) outs = [] outs_by_path = {} @@ -1182,17 +1196,17 @@ class Project(object): return G, G_active - def pipelines(self): + def pipelines(self, from_directory=None): import networkx as nx - G, G_active = self.graph() + G, G_active = self.graph(from_directory) return [ G.subgraph(c).copy() for c in nx.weakly_connected_components(G) ] - def stages(self): + def stages(self, from_directory=None): """ Walks down the root directory looking for Dvcfiles, skipping the directories that are related with @@ -1203,9 +1217,12 @@ class Project(object): """ from dvc.stage import Stage + if not from_directory: + from_directory = self.root_dir + stages = [] outs = [] - for root, dirs, files in os.walk(self.root_dir): + for root, dirs, files in os.walk(from_directory): for fname in files: path = os.path.join(root, fname) if not Stage.is_stage_file(path): @@ -1228,11 +1245,11 @@ class Project(object): return stages - def active_stages(self): + def active_stages(self, from_directory=None): import networkx as nx stages = [] - for G in self.pipelines(): + for G in self.pipelines(from_directory): stages.extend(list(nx.get_node_attributes(G, 'stage').values())) return stages diff --git a/tests/test_data_cloud.py b/tests/test_data_cloud.py index 2ba6ce17c..d13eb4040 100644 --- a/tests/test_data_cloud.py +++ b/tests/test_data_cloud.py @@ -15,6 +15,7 @@ from dvc.config import Config from dvc.data_cloud import (DataCloud, RemoteS3, RemoteGS, RemoteAzure, RemoteLOCAL, RemoteSSH, RemoteHDFS, RemoteHTTP) from dvc.remote.base import STATUS_OK, STATUS_NEW, STATUS_DELETED +from dvc.utils import file_md5 from tests.basic_env import TestDvc @@ -597,3 +598,89 @@ class TestWarnOnOutdatedStage(TestDvc): self.color_patch.start() self._test() self.color_patch.stop() + + +class TestRecursiveSyncOperations(TestDataCloudBase): + def main(self, args): + ret = main(args) + self.assertEqual(ret, 0) + + def _get_url(self): + self.dname = get_local_url() + return self.dname + + def _should_test(self): + return True + + def _get_cloud_class(self): + return RemoteLOCAL + + def _prepare_repo(self): + self.main(['remote', + 'add', + '-d', + TEST_REMOTE, + self.cloud._cloud.cache_dir]) + self.create(self.DATA, self.DATA_CONTENTS) + self.create(self.DATA_SUB, self.DATA_SUB_CONTENTS) + + self.dvc.add(self.DATA) + self.dvc.add(self.DATA_SUB_CONTENTS) + + def _remove_local_data_files(self): + os.remove(self.DATA) + os.remove(self.DATA_SUB) + + def _test_recursive_pull(self): + self._remove_local_data_files() + self._clear_local_cache() + + self.assertFalse(os.path.exists(self.DATA)) + self.assertFalse(os.path.exists(self.DATA_SUB)) + + self.main(['pull', '-R', self.DATA_DIR]) + + self.assertTrue(os.path.exists(self.DATA)) + self.assertTrue(os.path.exists(self.DATA_SUB)) + + def _clear_local_cache(self): + shutil.rmtree(self.dvc.cache.local.cache_dir) + + def _test_recursive_fetch(self, data_md5, data_sub_md5): + self._clear_local_cache() + + local_cache_data_path = self.dvc.cache.local.get(data_md5) + local_cache_data_sub_path = self.dvc.cache.local.get(data_sub_md5) + + self.assertFalse(os.path.exists(local_cache_data_path)) + self.assertFalse(os.path.exists(local_cache_data_sub_path)) + + self.main(['fetch', '-R', self.DATA_DIR]) + + self.assertTrue(os.path.exists(local_cache_data_path)) + self.assertTrue(os.path.exists(local_cache_data_sub_path)) + + def _test_recursive_push(self, data_md5, data_sub_md5): + cloud_data_path = self.cloud._cloud.get(data_md5) + cloud_data_sub_path = self.cloud._cloud.get(data_sub_md5) + + self.assertFalse(os.path.exists(cloud_data_path)) + self.assertFalse(os.path.exists(cloud_data_sub_path)) + + self.main(['push', '-R', self.DATA_DIR]) + + self.assertTrue(os.path.exists(cloud_data_path)) + self.assertTrue(os.path.exists(cloud_data_sub_path)) + + def test(self): + self._setup_cloud() + self._prepare_repo() + + data_md5 = file_md5(self.DATA)[0] + data_sub_md5 = file_md5(self.DATA_SUB)[0] + + self._test_recursive_push(data_md5, data_sub_md5) + + self._test_recursive_fetch(data_md5, data_sub_md5) + + self._test_recursive_pull()