Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-10 15:28:34 +01:00
Code Issues Packages Projects Releases Wiki Activity

dvc: support deps/outputs that are relative to specific remote

Browse Source 
Fixes #748

Signed-off-by: Ruslan Kuprieiev <[email protected]>
This commit is contained in:
Ruslan Kuprieiev
2018-06-09 02:03:02 +03:00
parent fecc38e602
commit a734e07776
14 changed files with 106 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/dependency/__init__.py
+25 -7
View File
@@ -1,18 +1,30 @@
import schema
try:
from urlparse import urlparse
except ImportError:
from urllib.parse import urlparse
from dvc.exceptions import DvcException
from dvc.config import Config
from dvc.dependency.base import DependencyBase
from dvc.dependency.s3 import DependencyS3
from dvc.dependency.gs import DependencyGS
from dvc.dependency.local import DependencyLOCAL
from dvc.remote import Remote
from dvc.remote.local import RemoteLOCAL
from dvc.remote.s3 import RemoteS3
from dvc.remote.gs import RemoteGS
from dvc.remote.ssh import RemoteSSH
DEPS = [DependencyS3, DependencyGS, DependencyLOCAL]
DEP_MAP = {'': DependencyLOCAL,
's3': DependencyS3,
'gs': DependencyGS}
SCHEMA = {
DependencyBase.PARAM_PATH: str,
schema.Optional(RemoteLOCAL.PARAM_MD5): schema.Or(str, None),
@@ -20,23 +32,29 @@ SCHEMA = {
}
def _get(path):
def _get(stage, p, info):
parsed = urlparse(p)
if parsed.scheme == 'remote':
sect = stage.project.config._config[Config.SECTION_REMOTE_FMT.format(parsed.netloc)]
remote = Remote(stage.project, sect)
return DEP_MAP[remote.scheme](stage, p, info, remote=remote)
for d in DEPS:
if d.supported(path):
return d
raise DvcException('Dependency \'{}\' is not supported'.format(path))
if d.supported(p):
return d(stage, p, info)
raise DvcException('Dependency \'{}\' is not supported'.format(p))
def loadd_from(stage, d_list):
ret = []
for d in d_list:
p = d.pop(DependencyBase.PARAM_PATH)
ret.append(_get(p)(stage, p, d))
ret.append(_get(stage, p, d))
return ret
def loads_from(stage, s_list):
ret = []
for s in s_list:
ret.append(_get(s)(stage, s, {}))
ret.append(_get(stage, s, {}))
return ret
dvc/dependency/gs.py
+10 -4
View File
@@ -1,3 +1,5 @@
import posixpath
try:
from urlparse import urlparse
except ImportError:
@@ -12,10 +14,14 @@ from dvc.config import Config
class DependencyGS(DependencyS3):
REGEX = DataCloudGCP.REGEX
def __init__(self, stage, path, info=None):
def __init__(self, stage, path, info=None, remote=None):
super(DependencyGS, self).__init__(stage, path)
self.info = info
self.remote = RemoteGS(stage.project, {})
self.remote = remote if remote else RemoteGS(stage.project, {})
bucket = remote.bucket if remote else urlparse(path).netloc
key = urlparse(path).path.lstrip('/')
if remote:
key = posixpath.join(remote.prefix, key)
self.path_info = {'scheme': 'gs',
'bucket': urlparse(path).netloc,
'key': urlparse(path).path.lstrip('/')}
'bucket': bucket,
'key': key}
dvc/dependency/local.py
+2 -2
View File
@@ -17,7 +17,7 @@ class DependencyLOCAL(DependencyBase):
DoesNotExistError = DependencyDoesNotExistError
IsNotFileOrDirError = DependencyIsNotFileOrDirError
def __init__(self, stage, path, info=None):
def __init__(self, stage, path, info=None, remote=None):
self.stage = stage
self.project = stage.project
if not os.path.isabs(path):
@@ -25,7 +25,7 @@ class DependencyLOCAL(DependencyBase):
path = os.path.join(stage.cwd, path)
self.path = os.path.normpath(path)
self.info = info
self.remote = RemoteLOCAL(stage.project, {})
self.remote = remote if remote != None else RemoteLOCAL(stage.project, {})
self.path_info = {'scheme': 'local',
'path': self.path}
dvc/dependency/s3.py
+11 -4
View File
@@ -1,3 +1,5 @@
import posixpath
try:
from urlparse import urlparse
except ImportError:
@@ -12,13 +14,18 @@ from dvc.config import Config
class DependencyS3(DependencyBase):
REGEX = DataCloudAWS.REGEX
def __init__(self, stage, path, info=None):
def __init__(self, stage, path, info=None, remote=None):
super(DependencyS3, self).__init__(stage, path)
self.info = info
self.remote = RemoteS3(stage.project, {})
self.remote = remote if remote else RemoteS3(stage.project, {})
bucket = remote.bucket if remote else urlparse(path).netloc
key = urlparse(path).path.lstrip('/')
if remote:
key = posixpath.join(remote.prefix, key)
self.path_info = {'scheme': 's3',
'bucket': urlparse(path).netloc,
'key': urlparse(path).path.lstrip('/')}
'bucket': bucket,
'key': key}
def changed(self):
return self.info != self.remote.save_info(self.path_info)
dvc/output/__init__.py
+20 -8
View File
@@ -1,25 +1,37 @@
import schema
from dvc.exceptions import DvcException
from dvc.config import Config
from dvc.dependency import SCHEMA as DEP_SCHEMA
from dvc.dependency import SCHEMA, urlparse
from dvc.dependency.base import DependencyBase
from dvc.output.s3 import OutputS3
from dvc.output.gs import OutputGS
from dvc.output.local import OutputLOCAL
from dvc.remote import Remote
OUTS = [OutputS3, OutputGS, OutputLOCAL]
SCHEMA = DEP_SCHEMA
OUTS_MAP = {'s3': OutputS3,
'gs': OutputGS,
'': OutputLOCAL}
SCHEMA[schema.Optional(OutputLOCAL.PARAM_CACHE)] = bool
def _get(path):
def _get(stage, p, info, cache):
parsed = urlparse(p)
if parsed.scheme == 'remote':
sect = stage.project.config._config[Config.SECTION_REMOTE_FMT.format(parsed.netloc)]
remote = Remote(stage.project, sect)
return OUTS_MAP[remote.scheme](stage, p, info, cache=cache, remote=remote)
for o in OUTS:
if o.supported(path):
return o
raise DvcException('Output \'{}\' is not supported'.format(path))
if o.supported(p):
return o(stage, p, info, cache=cache, remote=None)
raise DvcException('Output \'{}\' is not supported'.format(p))
def loadd_from(stage, d_list):
@@ -27,12 +39,12 @@ def loadd_from(stage, d_list):
for d in d_list:
p = d.pop(DependencyBase.PARAM_PATH)
cache = d.pop(OutputLOCAL.PARAM_CACHE, True)
ret.append(_get(p)(stage, p, info=d, cache=cache))
ret.append(_get(stage, p, info=d, cache=cache))
return ret
def loads_from(stage, s_list, use_cache=True):
ret = []
for s in s_list:
ret.append(_get(s)(stage, s, info={}, cache=use_cache))
ret.append(_get(stage, s, info={}, cache=use_cache))
return ret
dvc/output/gs.py
+2 -2
View File
@@ -5,8 +5,8 @@ from dvc.exceptions import DvcException
class OutputGS(DependencyGS):
PARAM_CACHE = 'cache'
def __init__(self, stage, path, info=None, cache=True):
super(OutputGS, self).__init__(stage, path, info)
def __init__(self, stage, path, info=None, remote=None, cache=True):
super(OutputGS, self).__init__(stage, path, info, remote=remote)
self.use_cache = cache
if cache and self.project.cache.gs == None:
raise DvcException("No cache location setup for \'gs\' outputs.")
dvc/output/local.py
+2 -2
View File
@@ -16,8 +16,8 @@ class OutputLOCAL(DependencyLOCAL):
DoesNotExistError = OutputDoesNotExistError
IsNotFileOrDirError = OutputIsNotFileOrDirError
def __init__(self, stage, path, info=None, cache=True):
super(OutputLOCAL, self).__init__(stage, path, info)
def __init__(self, stage, path, info=None, remote=None, cache=True):
super(OutputLOCAL, self).__init__(stage, path, info, remote=remote)
self.use_cache = cache
@property
dvc/output/s3.py
+2 -2
View File
@@ -5,8 +5,8 @@ from dvc.exceptions import DvcException
class OutputS3(DependencyS3):
PARAM_CACHE = 'cache'
def __init__(self, stage, path, info=None, cache=True):
super(OutputS3, self).__init__(stage, path, info)
def __init__(self, stage, path, info=None, remote=None, cache=True):
super(OutputS3, self).__init__(stage, path, info, remote=remote)
self.use_cache = cache
if cache and self.project.cache.s3 == None:
raise DvcException("No cache location setup for \'s3\' outputs.")
dvc/remote/gs.py
+1
View File
@@ -11,6 +11,7 @@ from dvc.config import Config
class RemoteGS(RemoteBase):
scheme = 'gs'
REGEX = r'^gs://(?P<path>.*)$'
PARAM_ETAG = 'etag'
dvc/remote/local.py
+3 -2
View File
@@ -12,6 +12,7 @@ from dvc.config import Config
class RemoteLOCAL(RemoteBase):
scheme = ''
REGEX = r'^(?P<path>(/+|.:\\+).*)$'
PARAM_MD5 = State.PARAM_MD5
@@ -26,10 +27,10 @@ class RemoteLOCAL(RemoteBase):
def __init__(self, project, config):
self.project = project
self.link_state = project.link_state
self.cache_dir = config.get(Config.SECTION_REMOTE_URL, os.path.abspath(os.sep))
self.cache_dir = config.get(Config.SECTION_REMOTE_URL, None)
self.cache_type = config.get(Config.SECTION_CACHE_TYPE, None)
if not os.path.exists(self.cache_dir):
if self.cache_dir != None and not os.path.exists(self.cache_dir):
os.mkdir(self.cache_dir)
self.state = State(self.cache_dir)
dvc/remote/s3.py
+1
View File
@@ -11,6 +11,7 @@ from dvc.remote.base import RemoteBase
class RemoteS3(RemoteBase):
scheme = 's3'
REGEX = r'^s3://(?P<path>.*)$'
PARAM_ETAG = 'etag'
dvc/remote/ssh.py
+1
View File
@@ -2,6 +2,7 @@ from dvc.remote.base import RemoteBase
class RemoteSSH(RemoteBase):
scheme = 'ssh'
#NOTE: ~/ paths are temporarily forbidden
REGEX = r'^(?P<user>.*)@(?P<host>.*):(?P<path>/+.*)$'
pass
dvc/state.py
+11 -4
View File
@@ -48,10 +48,14 @@ class State(object):
def __init__(self, dvc_dir):
self.dvc_dir = dvc_dir
self.state_file = os.path.join(dvc_dir, self.STATE_FILE)
self._db = self.load()
self._lock = threading.Lock()
self._lock_file = Lock(dvc_dir, self.STATE_LOCK_FILE)
if dvc_dir:
self.state_file = os.path.join(dvc_dir, self.STATE_FILE)
self._lock_file = Lock(dvc_dir, self.STATE_LOCK_FILE)
else:
self.state_file = None
self._lock_file = threading.Lock()
self._db = self.load()
@staticmethod
def init(dvc_dir):
@@ -88,13 +92,16 @@ class State(object):
return actual.split('.')[0] != md5.split('.')[0]
def load(self):
if not os.path.isfile(self.state_file):
if not self.state_file or not os.path.isfile(self.state_file):
return {}
with open(self.state_file, 'r') as fd:
return json.load(fd)
def dump(self):
if not self.state_file:
return
with open(self.state_file, 'w+') as fd:
json.dump(self._db, fd)
tests/test_repro.py
+15 -4
View File
@@ -232,18 +232,29 @@ class TestReproExternalBase(TestDvc):
ret = main(['remote', 'add', 'myrepo', cache])
self.assertEqual(ret, 0)
remote_name = 'myremote'
remote_key = str(uuid.uuid4())
remote = self.scheme + '://' + self.bucket + '/' + remote_key
ret = main(['remote', 'add', remote_name, remote])
self.assertEqual(ret, 0)
self.dvc = Project('.')
foo_key = str(uuid.uuid4()) + '/' + self.FOO
bar_key = str(uuid.uuid4()) + '/' + self.BAR
foo_key = remote_key + '/' + self.FOO
bar_key = remote_key + '/' + self.BAR
foo_path = self.scheme + '://' + self.bucket + '/' + foo_key
bar_path = self.scheme + '://' + self.bucket + '/' + bar_key
# Using both plain and remote notation
out_foo_path = 'remote://' + remote_name + '/' + self.FOO
out_bar_path = bar_path
self.write(self.bucket, foo_key, 'foo')
stage = self.dvc.run(outs=[bar_path],
deps=[foo_path],
stage = self.dvc.run(outs=[out_bar_path],
deps=[out_foo_path],
cmd='{} {} {}'.format(self.cmd, foo_path, bar_path))
self.write(self.bucket, foo_key, 'bar')