Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-02 07:18:25 +01:00
Code Issues Packages Projects Releases Wiki Activity

project: support recursive sync operations

Browse Source
This commit is contained in:
pared
2019-01-10 18:02:23 +01:00
parent a15f68a20d
commit 6f7e2095d3
4 changed files with 145 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/cli.py
+15
View File
@@ -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
dvc/command/data_sync.py
+6 -3
View File
@@ -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
dvc/project.py
+37 -20
View File
@@ -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
tests/test_data_cloud.py
+87
View File
@@ -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()