mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-12 07:18:23 +01:00
RFC: dependencies/outputs: unify classes
There is a lot of duplication in current Output/Dependency classes, that was left "just in case" while developing different schemes. Since our code base is now getting more and more stable and we now know what we need to add support for any new dep/out type in the future, we can now safely unify most of the code. Signed-off-by: Ruslan Kuprieiev <[email protected]>
This commit is contained in:
+1
-5
@@ -1,11 +1,7 @@
|
||||
import os
|
||||
|
||||
try:
|
||||
from urlparse import urlparse
|
||||
except ImportError:
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import dvc.logger as logger
|
||||
from dvc.utils import urlparse
|
||||
from dvc.exceptions import DvcException
|
||||
from dvc.command.base import CmdBase
|
||||
|
||||
|
||||
@@ -1,13 +1,10 @@
|
||||
import schema
|
||||
|
||||
try:
|
||||
from urlparse import urlparse
|
||||
except ImportError:
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from dvc.config import Config
|
||||
from dvc.utils import urlparse
|
||||
|
||||
from dvc.dependency.base import DependencyBase
|
||||
import dvc.output as output
|
||||
from dvc.output.base import OutputBase
|
||||
from dvc.dependency.s3 import DependencyS3
|
||||
from dvc.dependency.gs import DependencyGS
|
||||
from dvc.dependency.local import DependencyLOCAL
|
||||
@@ -16,9 +13,6 @@ from dvc.dependency.ssh import DependencySSH
|
||||
from dvc.dependency.http import DependencyHTTP
|
||||
|
||||
from dvc.remote import Remote
|
||||
from dvc.remote.local import RemoteLOCAL
|
||||
from dvc.remote.s3 import RemoteS3
|
||||
from dvc.remote.hdfs import RemoteHDFS
|
||||
|
||||
DEPS = [
|
||||
DependencyGS,
|
||||
@@ -30,7 +24,7 @@ DEPS = [
|
||||
]
|
||||
|
||||
DEP_MAP = {
|
||||
'': DependencyLOCAL,
|
||||
'local': DependencyLOCAL,
|
||||
'ssh': DependencySSH,
|
||||
's3': DependencyS3,
|
||||
'gs': DependencyGS,
|
||||
@@ -39,13 +33,9 @@ DEP_MAP = {
|
||||
'https': DependencyHTTP,
|
||||
}
|
||||
|
||||
# We are skipping RemoteHTTP.PARAM_CHECKSUM because is the same as RemoteS3
|
||||
SCHEMA = {
|
||||
DependencyBase.PARAM_PATH: str,
|
||||
schema.Optional(RemoteLOCAL.PARAM_CHECKSUM): schema.Or(str, None),
|
||||
schema.Optional(RemoteS3.PARAM_CHECKSUM): schema.Or(str, None),
|
||||
schema.Optional(RemoteHDFS.PARAM_CHECKSUM): schema.Or(str, None),
|
||||
}
|
||||
SCHEMA = output.SCHEMA.copy()
|
||||
del SCHEMA[schema.Optional(OutputBase.PARAM_CACHE)]
|
||||
del SCHEMA[schema.Optional(OutputBase.PARAM_METRIC)]
|
||||
|
||||
|
||||
def _get(stage, p, info):
|
||||
@@ -65,7 +55,7 @@ def _get(stage, p, info):
|
||||
def loadd_from(stage, d_list):
|
||||
ret = []
|
||||
for d in d_list:
|
||||
p = d.pop(DependencyBase.PARAM_PATH)
|
||||
p = d.pop(OutputBase.PARAM_PATH)
|
||||
ret.append(_get(stage, p, d))
|
||||
return ret
|
||||
|
||||
|
||||
+3
-69
@@ -1,5 +1,3 @@
|
||||
import re
|
||||
|
||||
from dvc.exceptions import DvcException
|
||||
|
||||
|
||||
@@ -16,71 +14,7 @@ class DependencyIsNotFileOrDirError(DvcException):
|
||||
|
||||
|
||||
class DependencyBase(object):
|
||||
REGEX = None
|
||||
IS_DEPENDENCY = True
|
||||
|
||||
PARAM_PATH = 'path'
|
||||
|
||||
def __init__(self, stage, path, info=None):
|
||||
self.stage = stage
|
||||
self.project = stage.project
|
||||
self.url = path
|
||||
self.info = info
|
||||
|
||||
def __repr__(self):
|
||||
return "{class_name}: '{url}'".format(
|
||||
class_name=type(self).__name__,
|
||||
url=(self.url or 'No url')
|
||||
)
|
||||
|
||||
def __str__(self):
|
||||
return self.url
|
||||
|
||||
@classmethod
|
||||
def match(cls, url):
|
||||
return re.match(cls.REGEX, url)
|
||||
|
||||
def group(self, name):
|
||||
match = self.match(self.url)
|
||||
if not match:
|
||||
return None
|
||||
return match.group(name)
|
||||
|
||||
@classmethod
|
||||
def supported(cls, url):
|
||||
return cls.match(url) is not None
|
||||
|
||||
@property
|
||||
def scheme(self):
|
||||
return self.path_info['scheme']
|
||||
|
||||
@property
|
||||
def path(self):
|
||||
return self.path_info['path']
|
||||
|
||||
@property
|
||||
def sep(self):
|
||||
return '/'
|
||||
|
||||
@property
|
||||
def exists(self):
|
||||
return self.remote.exists(self.path_info)
|
||||
|
||||
def changed(self):
|
||||
raise NotImplementedError
|
||||
|
||||
def status(self):
|
||||
if self.changed():
|
||||
# FIXME better msgs
|
||||
return {str(self): 'changed'}
|
||||
return {}
|
||||
|
||||
def save(self):
|
||||
raise NotImplementedError
|
||||
|
||||
def dumpd(self):
|
||||
ret = self.info.copy()
|
||||
ret[self.PARAM_PATH] = self.url
|
||||
return ret
|
||||
|
||||
def download(self, to_info):
|
||||
self.remote.download([self.path_info], [to_info])
|
||||
DoesNotExistError = DependencyDoesNotExistError
|
||||
IsNotFileOrDirError = DependencyIsNotFileOrDirError
|
||||
|
||||
+4
-22
@@ -1,24 +1,6 @@
|
||||
import posixpath
|
||||
|
||||
try:
|
||||
from urlparse import urlparse
|
||||
except ImportError:
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from dvc.dependency.s3 import DependencyS3
|
||||
from dvc.remote.gs import RemoteGS
|
||||
from dvc.output.gs import OutputGS
|
||||
from dvc.dependency.base import DependencyBase
|
||||
|
||||
|
||||
class DependencyGS(DependencyS3):
|
||||
REGEX = RemoteGS.REGEX
|
||||
|
||||
def __init__(self, stage, path, info=None, remote=None):
|
||||
super(DependencyGS, self).__init__(stage, path, info=info)
|
||||
self.remote = remote if remote else RemoteGS(stage.project, {})
|
||||
bucket = remote.bucket if remote else urlparse(path).netloc
|
||||
path = urlparse(path).path.lstrip('/')
|
||||
if remote:
|
||||
path = posixpath.join(remote.prefix, path)
|
||||
self.path_info = {'scheme': 'gs',
|
||||
'bucket': bucket,
|
||||
'path': path}
|
||||
class DependencyGS(DependencyBase, OutputGS):
|
||||
pass
|
||||
|
||||
+3
-29
@@ -1,32 +1,6 @@
|
||||
import posixpath
|
||||
|
||||
try:
|
||||
from urlparse import urlparse
|
||||
except ImportError:
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from dvc.output.hdfs import OutputHDFS
|
||||
from dvc.dependency.base import DependencyBase
|
||||
from dvc.remote.hdfs import RemoteHDFS
|
||||
|
||||
|
||||
class DependencyHDFS(DependencyBase):
|
||||
REGEX = RemoteHDFS.REGEX
|
||||
|
||||
def __init__(self, stage, path, info=None, remote=None):
|
||||
super(DependencyHDFS, self).__init__(stage, path, info=info)
|
||||
self.remote = remote if remote else RemoteHDFS(stage.project, {})
|
||||
if remote:
|
||||
path = posixpath.join(remote.url, urlparse(path).path.lstrip('/'))
|
||||
user = remote.user if remote else self.group('user')
|
||||
self.path_info = {'scheme': 'hdfs',
|
||||
'user': user,
|
||||
'path': path}
|
||||
|
||||
def changed(self):
|
||||
if not self.exists:
|
||||
return True
|
||||
|
||||
return self.info != self.remote.save_info(self.path_info)
|
||||
|
||||
def save(self):
|
||||
self.info = self.remote.save_info(self.path_info)
|
||||
class DependencyHDFS(DependencyBase, OutputHDFS):
|
||||
pass
|
||||
|
||||
+9
-21
@@ -1,20 +1,17 @@
|
||||
try:
|
||||
from urlparse import urlparse, urljoin
|
||||
except ImportError:
|
||||
from urllib.parse import urlparse, urljoin
|
||||
|
||||
from dvc.dependency.base import DependencyBase
|
||||
from dvc.utils import urlparse, urljoin
|
||||
from dvc.output.base import OutputBase
|
||||
from dvc.remote.http import RemoteHTTP
|
||||
from dvc.dependency.base import DependencyBase
|
||||
|
||||
|
||||
class DependencyHTTP(DependencyBase):
|
||||
REGEX = RemoteHTTP.REGEX
|
||||
class DependencyHTTP(DependencyBase, OutputBase):
|
||||
REMOTE = RemoteHTTP
|
||||
|
||||
def __init__(self, stage, path, info=None, remote=None):
|
||||
super(DependencyHTTP, self).__init__(stage, path, info=info)
|
||||
|
||||
self.remote = remote or RemoteHTTP(stage.project, {})
|
||||
|
||||
super(DependencyHTTP, self).__init__(stage,
|
||||
path,
|
||||
info=info,
|
||||
remote=remote)
|
||||
if path.startswith('remote'):
|
||||
path = urljoin(self.remote.cache_dir, urlparse(path).path)
|
||||
|
||||
@@ -22,12 +19,3 @@ class DependencyHTTP(DependencyBase):
|
||||
'scheme': urlparse(path).scheme,
|
||||
'path': path,
|
||||
}
|
||||
|
||||
def save(self):
|
||||
self.info = self.remote.save_info(self.path_info)
|
||||
|
||||
def changed(self):
|
||||
if not self.exists:
|
||||
return True
|
||||
|
||||
return self.info != self.remote.save_info(self.path_info)
|
||||
|
||||
+3
-87
@@ -1,90 +1,6 @@
|
||||
import os
|
||||
|
||||
try:
|
||||
from urlparse import urlparse
|
||||
except ImportError:
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import dvc.logger as logger
|
||||
from dvc.dependency.base import DependencyBase
|
||||
from dvc.dependency.base import DependencyDoesNotExistError
|
||||
from dvc.dependency.base import DependencyIsNotFileOrDirError
|
||||
from dvc.remote.local import RemoteLOCAL
|
||||
from dvc.output.local import OutputLOCAL
|
||||
|
||||
|
||||
class DependencyLOCAL(DependencyBase):
|
||||
REGEX = r'^(?P<path>.*)$'
|
||||
|
||||
DoesNotExistError = DependencyDoesNotExistError
|
||||
IsNotFileOrDirError = DependencyIsNotFileOrDirError
|
||||
|
||||
def __init__(self, stage, path, info=None, remote=None):
|
||||
super(DependencyLOCAL, self).__init__(stage, path, info)
|
||||
if remote is not None:
|
||||
self.remote = remote
|
||||
else:
|
||||
self.remote = RemoteLOCAL(stage.project, {})
|
||||
|
||||
if remote:
|
||||
p = os.path.join(remote.prefix,
|
||||
urlparse(self.url).path.lstrip('/'))
|
||||
else:
|
||||
p = path
|
||||
|
||||
if not os.path.isabs(p):
|
||||
p = self.remote.to_ospath(p)
|
||||
p = os.path.join(stage.cwd, p)
|
||||
p = os.path.abspath(os.path.normpath(p))
|
||||
|
||||
self.path_info = {'scheme': 'local',
|
||||
'path': p}
|
||||
|
||||
def __str__(self):
|
||||
return self.rel_path
|
||||
|
||||
@property
|
||||
def is_local(self):
|
||||
return (urlparse(self.url).scheme != 'remote'
|
||||
and not os.path.isabs(self.url))
|
||||
|
||||
@property
|
||||
def sep(self):
|
||||
return os.sep
|
||||
|
||||
@property
|
||||
def rel_path(self):
|
||||
return os.path.relpath(self.path)
|
||||
|
||||
def changed(self):
|
||||
if not self.exists:
|
||||
return True
|
||||
|
||||
info = self.remote.save_info(self.path_info)
|
||||
|
||||
return self.info != info
|
||||
|
||||
def save(self):
|
||||
if not self.exists:
|
||||
raise self.DoesNotExistError(self.rel_path)
|
||||
|
||||
if not os.path.isfile(self.path) \
|
||||
and not os.path.isdir(self.path): # pragma: no cover
|
||||
raise self.IsNotFileOrDirError(self.rel_path)
|
||||
|
||||
if (os.path.isfile(self.path) and os.path.getsize(self.path) == 0) or \
|
||||
(os.path.isdir(self.path) and len(os.listdir(self.path)) == 0):
|
||||
msg = "file/directory '{}' is empty.".format(self.rel_path)
|
||||
logger.warning(msg)
|
||||
|
||||
self.info = self.remote.save_info(self.path_info)
|
||||
|
||||
def dumpd(self):
|
||||
if self.is_local:
|
||||
path = self.remote.unixpath(os.path.relpath(self.path,
|
||||
self.stage.cwd))
|
||||
else:
|
||||
path = self.url
|
||||
|
||||
info = self.info.copy()
|
||||
info[self.PARAM_PATH] = path
|
||||
return info
|
||||
class DependencyLOCAL(DependencyBase, OutputLOCAL):
|
||||
pass
|
||||
|
||||
+3
-31
@@ -1,34 +1,6 @@
|
||||
import posixpath
|
||||
|
||||
try:
|
||||
from urlparse import urlparse
|
||||
except ImportError:
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from dvc.output.s3 import OutputS3
|
||||
from dvc.dependency.base import DependencyBase
|
||||
from dvc.remote.s3 import RemoteS3
|
||||
|
||||
|
||||
class DependencyS3(DependencyBase):
|
||||
REGEX = RemoteS3.REGEX
|
||||
|
||||
def __init__(self, stage, path, info=None, remote=None):
|
||||
super(DependencyS3, self).__init__(stage, path, info=info)
|
||||
self.remote = remote or RemoteS3(stage.project, {})
|
||||
|
||||
bucket = remote.bucket if remote else urlparse(path).netloc
|
||||
path = urlparse(path).path.lstrip('/')
|
||||
if remote:
|
||||
path = posixpath.join(remote.prefix, path)
|
||||
self.path_info = {'scheme': 's3',
|
||||
'bucket': bucket,
|
||||
'path': path}
|
||||
|
||||
def changed(self):
|
||||
if not self.exists:
|
||||
return True
|
||||
|
||||
return self.info != self.remote.save_info(self.path_info)
|
||||
|
||||
def save(self):
|
||||
self.info = self.remote.save_info(self.path_info)
|
||||
class DependencyS3(DependencyBase, OutputS3):
|
||||
pass
|
||||
|
||||
+3
-38
@@ -1,41 +1,6 @@
|
||||
import posixpath
|
||||
|
||||
try:
|
||||
from urlparse import urlparse
|
||||
except ImportError:
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from dvc.output.ssh import OutputSSH
|
||||
from dvc.dependency.base import DependencyBase
|
||||
from dvc.remote.ssh import RemoteSSH
|
||||
|
||||
|
||||
class DependencySSH(DependencyBase):
|
||||
REGEX = RemoteSSH.REGEX
|
||||
|
||||
def __init__(self, stage, path, info=None, remote=None):
|
||||
super(DependencySSH, self).__init__(stage, path, info=info)
|
||||
self.remote = remote if remote else RemoteSSH(stage.project, {})
|
||||
|
||||
host = remote.host if remote else self.group('host')
|
||||
port = remote.port if remote else RemoteSSH.DEFAULT_PORT
|
||||
user = remote.user if remote else self.group('user')
|
||||
if remote:
|
||||
path = posixpath.join(remote.prefix,
|
||||
urlparse(path).path.lstrip('/'))
|
||||
else:
|
||||
path = self.match(self.url).group('path')
|
||||
|
||||
self.path_info = {'scheme': 'ssh',
|
||||
'host': host,
|
||||
'port': port,
|
||||
'user': user,
|
||||
'path': path}
|
||||
|
||||
def changed(self):
|
||||
if not self.exists:
|
||||
return True
|
||||
|
||||
return self.info != self.remote.save_info(self.path_info)
|
||||
|
||||
def save(self):
|
||||
self.info = self.remote.save_info(self.path_info)
|
||||
class DependencySSH(DependencyBase, OutputSSH):
|
||||
pass
|
||||
|
||||
+18
-9
@@ -1,9 +1,9 @@
|
||||
import schema
|
||||
|
||||
from dvc.config import Config
|
||||
from dvc.utils import urlparse
|
||||
|
||||
from dvc.dependency import SCHEMA, urlparse
|
||||
from dvc.dependency.base import DependencyBase
|
||||
from dvc.output.base import OutputBase
|
||||
from dvc.output.s3 import OutputS3
|
||||
from dvc.output.gs import OutputGS
|
||||
from dvc.output.local import OutputLOCAL
|
||||
@@ -11,7 +11,9 @@ from dvc.output.hdfs import OutputHDFS
|
||||
from dvc.output.ssh import OutputSSH
|
||||
|
||||
from dvc.remote import Remote
|
||||
|
||||
from dvc.remote.s3 import RemoteS3
|
||||
from dvc.remote.hdfs import RemoteHDFS
|
||||
from dvc.remote.local import RemoteLOCAL
|
||||
|
||||
OUTS = [
|
||||
OutputHDFS,
|
||||
@@ -25,10 +27,17 @@ OUTS_MAP = {'hdfs': OutputHDFS,
|
||||
's3': OutputS3,
|
||||
'gs': OutputGS,
|
||||
'ssh': OutputSSH,
|
||||
'': OutputLOCAL}
|
||||
'local': OutputLOCAL}
|
||||
|
||||
SCHEMA[schema.Optional(OutputLOCAL.PARAM_CACHE)] = bool
|
||||
SCHEMA[schema.Optional(OutputLOCAL.PARAM_METRIC)] = OutputLOCAL.METRIC_SCHEMA
|
||||
# We are skipping RemoteHTTP.PARAM_CHECKSUM because is the same as RemoteS3
|
||||
SCHEMA = {
|
||||
OutputBase.PARAM_PATH: str,
|
||||
schema.Optional(RemoteLOCAL.PARAM_CHECKSUM): schema.Or(str, None),
|
||||
schema.Optional(RemoteS3.PARAM_CHECKSUM): schema.Or(str, None),
|
||||
schema.Optional(RemoteHDFS.PARAM_CHECKSUM): schema.Or(str, None),
|
||||
schema.Optional(OutputBase.PARAM_CACHE): bool,
|
||||
schema.Optional(OutputBase.PARAM_METRIC): OutputBase.METRIC_SCHEMA,
|
||||
}
|
||||
|
||||
|
||||
def _get(stage, p, info, cache, metric):
|
||||
@@ -53,9 +62,9 @@ def _get(stage, p, info, cache, metric):
|
||||
def loadd_from(stage, d_list):
|
||||
ret = []
|
||||
for d in d_list:
|
||||
p = d.pop(DependencyBase.PARAM_PATH)
|
||||
cache = d.pop(OutputLOCAL.PARAM_CACHE, True)
|
||||
metric = d.pop(OutputLOCAL.PARAM_METRIC, False)
|
||||
p = d.pop(OutputBase.PARAM_PATH)
|
||||
cache = d.pop(OutputBase.PARAM_CACHE, True)
|
||||
metric = d.pop(OutputBase.PARAM_METRIC, False)
|
||||
ret.append(_get(stage, p, info=d, cache=cache, metric=metric))
|
||||
return ret
|
||||
|
||||
|
||||
@@ -1,3 +1,6 @@
|
||||
import re
|
||||
from schema import Or, Optional
|
||||
|
||||
from dvc.exceptions import DvcException
|
||||
|
||||
|
||||
@@ -17,3 +20,153 @@ class OutputAlreadyTrackedError(DvcException):
|
||||
def __init__(self, path):
|
||||
msg = "output '{}' is already tracked by scm (e.g. git)".format(path)
|
||||
super(OutputAlreadyTrackedError, self).__init__(msg)
|
||||
|
||||
|
||||
class OutputBase(object):
|
||||
IS_DEPENDENCY = False
|
||||
|
||||
REMOTE = None
|
||||
|
||||
PARAM_PATH = 'path'
|
||||
PARAM_CACHE = 'cache'
|
||||
PARAM_METRIC = 'metric'
|
||||
PARAM_METRIC_TYPE = 'type'
|
||||
PARAM_METRIC_XPATH = 'xpath'
|
||||
|
||||
METRIC_SCHEMA = Or(None, bool,
|
||||
{Optional(PARAM_METRIC_TYPE): Or(str, None),
|
||||
Optional(PARAM_METRIC_XPATH): Or(str, None)})
|
||||
|
||||
DoesNotExistError = OutputDoesNotExistError
|
||||
IsNotFileOrDirError = OutputIsNotFileOrDirError
|
||||
|
||||
def __init__(self,
|
||||
stage,
|
||||
path,
|
||||
info=None,
|
||||
remote=None,
|
||||
cache=True,
|
||||
metric=False):
|
||||
self.stage = stage
|
||||
self.project = stage.project
|
||||
self.url = path
|
||||
self.info = info
|
||||
self.remote = remote or self.REMOTE(self.project, {})
|
||||
self.use_cache = False if self.IS_DEPENDENCY else cache
|
||||
self.metric = False if self.IS_DEPENDENCY else metric
|
||||
|
||||
if self.use_cache and getattr(self.project.cache,
|
||||
self.REMOTE.scheme) is None:
|
||||
raise DvcException("no cache location setup for '{}' outputs."
|
||||
.format(self.REMOTE.scheme))
|
||||
|
||||
def __repr__(self):
|
||||
return "{class_name}: '{url}'".format(
|
||||
class_name=type(self).__name__,
|
||||
url=(self.url or 'No url')
|
||||
)
|
||||
|
||||
def __str__(self):
|
||||
return self.url
|
||||
|
||||
@classmethod
|
||||
def match(cls, url):
|
||||
return re.match(cls.REMOTE.REGEX, url)
|
||||
|
||||
def group(self, name):
|
||||
match = self.match(self.url)
|
||||
if not match:
|
||||
return None
|
||||
return match.group(name)
|
||||
|
||||
@classmethod
|
||||
def supported(cls, url):
|
||||
return cls.match(url) is not None
|
||||
|
||||
@property
|
||||
def scheme(self):
|
||||
return self.REMOTE.scheme
|
||||
|
||||
@property
|
||||
def path(self):
|
||||
return self.path_info['path']
|
||||
|
||||
@property
|
||||
def sep(self):
|
||||
return '/'
|
||||
|
||||
@property
|
||||
def exists(self):
|
||||
return self.remote.exists(self.path_info)
|
||||
|
||||
def changed(self):
|
||||
if not self.exists:
|
||||
return True
|
||||
|
||||
if not self.use_cache:
|
||||
return self.info != self.remote.save_info(self.path_info)
|
||||
|
||||
return getattr(self.project.cache, self.scheme).changed(self.path_info,
|
||||
self.info)
|
||||
|
||||
def status(self):
|
||||
if self.changed():
|
||||
# FIXME better msgs
|
||||
return {str(self): 'changed'}
|
||||
return {}
|
||||
|
||||
def save(self):
|
||||
if not self.use_cache:
|
||||
self.info = self.remote.save_info(self.path_info)
|
||||
else:
|
||||
self.info = getattr(self.project.cache,
|
||||
self.scheme).save(self.path_info)
|
||||
|
||||
def dumpd(self):
|
||||
ret = self.info.copy()
|
||||
ret[self.PARAM_PATH] = self.url
|
||||
|
||||
if self.IS_DEPENDENCY:
|
||||
return ret
|
||||
|
||||
ret[self.PARAM_CACHE] = self.use_cache
|
||||
|
||||
if isinstance(self.metric, dict):
|
||||
if self.PARAM_METRIC_XPATH in self.metric \
|
||||
and not self.metric[self.PARAM_METRIC_XPATH]:
|
||||
del self.metric[self.PARAM_METRIC_XPATH]
|
||||
|
||||
ret[self.PARAM_METRIC] = self.metric
|
||||
|
||||
return ret
|
||||
|
||||
def download(self, to_info):
|
||||
self.remote.download([self.path_info], [to_info])
|
||||
|
||||
def checkout(self, force=False):
|
||||
if not self.use_cache:
|
||||
return
|
||||
|
||||
getattr(self.project.cache, self.scheme).checkout(self.path_info,
|
||||
self.info,
|
||||
force=force)
|
||||
|
||||
def remove(self, ignore_remove=False):
|
||||
self.remote.remove(self.path_info)
|
||||
if self.scheme != 'local':
|
||||
return
|
||||
|
||||
if ignore_remove and self.use_cache and self.is_local:
|
||||
self.project.scm.ignore_remove(self.path)
|
||||
|
||||
def move(self, out):
|
||||
if self.scheme == 'local' and self.use_cache and self.is_local:
|
||||
self.project.scm.ignore_remove(self.path)
|
||||
|
||||
self.remote.move(self.path_info, out.path_info)
|
||||
self.url = out.url
|
||||
self.path_info = out.path_info
|
||||
self.save()
|
||||
|
||||
if self.scheme == 'local' and self.use_cache and self.is_local:
|
||||
self.project.scm.ignore(self.path)
|
||||
|
||||
+4
-51
@@ -1,53 +1,6 @@
|
||||
from dvc.dependency.gs import DependencyGS
|
||||
from dvc.exceptions import DvcException
|
||||
from dvc.remote.gs import RemoteGS
|
||||
from dvc.output.s3 import OutputS3
|
||||
|
||||
|
||||
class OutputGS(DependencyGS):
|
||||
PARAM_CACHE = 'cache'
|
||||
PARAM_METRIC = 'metric'
|
||||
|
||||
def __init__(self,
|
||||
stage,
|
||||
path,
|
||||
info=None,
|
||||
remote=None,
|
||||
cache=True,
|
||||
metric=False):
|
||||
super(OutputGS, self).__init__(stage, path, info, remote=remote)
|
||||
self.use_cache = cache
|
||||
self.metric = metric
|
||||
if cache and self.project.cache.gs is None:
|
||||
raise DvcException("no cache location setup for 'gs' outputs.")
|
||||
|
||||
def dumpd(self):
|
||||
ret = super(OutputGS, self).dumpd()
|
||||
ret[self.PARAM_CACHE] = self.use_cache
|
||||
ret[self.PARAM_METRIC] = self.metric
|
||||
return ret
|
||||
|
||||
def changed(self):
|
||||
if super(OutputGS, self).changed():
|
||||
return True
|
||||
|
||||
if self.use_cache \
|
||||
and self.project.cache.gs.changed(self.path_info, self.info):
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def checkout(self, force=False):
|
||||
if not self.use_cache:
|
||||
return
|
||||
|
||||
self.project.cache.gs.checkout(self.path_info, self.info)
|
||||
|
||||
def save(self):
|
||||
super(OutputGS, self).save()
|
||||
|
||||
if not self.use_cache:
|
||||
return
|
||||
|
||||
self.info = self.project.cache.gs.save(self.path_info)
|
||||
|
||||
def remove(self, ignore_remove=False):
|
||||
self.remote.remove(self.path_info)
|
||||
class OutputGS(OutputS3):
|
||||
REMOTE = RemoteGS
|
||||
|
||||
+19
-43
@@ -1,10 +1,12 @@
|
||||
from dvc.dependency.hdfs import DependencyHDFS
|
||||
from dvc.exceptions import DvcException
|
||||
import posixpath
|
||||
|
||||
from dvc.utils import urlparse
|
||||
from dvc.output.base import OutputBase
|
||||
from dvc.remote.hdfs import RemoteHDFS
|
||||
|
||||
|
||||
class OutputHDFS(DependencyHDFS):
|
||||
PARAM_CACHE = 'cache'
|
||||
PARAM_METRIC = 'metric'
|
||||
class OutputHDFS(OutputBase):
|
||||
REMOTE = RemoteHDFS
|
||||
|
||||
def __init__(self,
|
||||
stage,
|
||||
@@ -13,41 +15,15 @@ class OutputHDFS(DependencyHDFS):
|
||||
remote=None,
|
||||
cache=True,
|
||||
metric=False):
|
||||
super(OutputHDFS, self).__init__(stage, path, info, remote=remote)
|
||||
self.use_cache = cache
|
||||
self.metric = metric
|
||||
if cache and self.project.cache.hdfs is None:
|
||||
raise DvcException("no cache location setup for 'hdfs' outputs.")
|
||||
|
||||
def dumpd(self):
|
||||
ret = super(OutputHDFS, self).dumpd()
|
||||
ret[self.PARAM_CACHE] = self.use_cache
|
||||
ret[self.PARAM_METRIC] = self.metric
|
||||
return ret
|
||||
|
||||
def changed(self):
|
||||
if super(OutputHDFS, self).changed():
|
||||
return True
|
||||
|
||||
if self.use_cache \
|
||||
and self.project.cache.hdfs.changed(self.path_info, self.info):
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def checkout(self, force=False):
|
||||
if not self.use_cache:
|
||||
return
|
||||
|
||||
self.project.cache.hdfs.checkout(self.path_info, self.info)
|
||||
|
||||
def save(self):
|
||||
super(OutputHDFS, self).save()
|
||||
|
||||
if not self.use_cache:
|
||||
return
|
||||
|
||||
self.info = self.project.cache.hdfs.save(self.path_info)
|
||||
|
||||
def remove(self, ignore_remove=False):
|
||||
self.remote.remove(self.path_info)
|
||||
super(OutputHDFS, self).__init__(stage,
|
||||
path,
|
||||
info=info,
|
||||
remote=remote,
|
||||
cache=cache,
|
||||
metric=metric)
|
||||
if remote:
|
||||
path = posixpath.join(remote.url, urlparse(path).path.lstrip('/'))
|
||||
user = remote.user if remote else self.group('user')
|
||||
self.path_info = {'scheme': 'hdfs',
|
||||
'user': user,
|
||||
'path': path}
|
||||
|
||||
+57
-66
@@ -1,26 +1,15 @@
|
||||
import os
|
||||
from schema import Optional, Or
|
||||
|
||||
import dvc.logger as logger
|
||||
from dvc.output.base import OutputDoesNotExistError, OutputIsNotFileOrDirError
|
||||
from dvc.output.base import OutputAlreadyTrackedError
|
||||
from dvc.dependency.local import DependencyLOCAL
|
||||
from dvc.exceptions import DvcException
|
||||
from dvc.utils import urlparse
|
||||
from dvc.istextfile import istextfile
|
||||
from dvc.exceptions import DvcException
|
||||
from dvc.remote.local import RemoteLOCAL
|
||||
from dvc.output.base import OutputBase, OutputAlreadyTrackedError
|
||||
|
||||
|
||||
class OutputLOCAL(DependencyLOCAL):
|
||||
PARAM_CACHE = 'cache'
|
||||
PARAM_METRIC = 'metric'
|
||||
PARAM_METRIC_TYPE = 'type'
|
||||
PARAM_METRIC_XPATH = 'xpath'
|
||||
|
||||
METRIC_SCHEMA = Or(None, bool,
|
||||
{Optional(PARAM_METRIC_TYPE): Or(str, None),
|
||||
Optional(PARAM_METRIC_XPATH): Or(str, None)})
|
||||
|
||||
DoesNotExistError = OutputDoesNotExistError
|
||||
IsNotFileOrDirError = OutputIsNotFileOrDirError
|
||||
class OutputLOCAL(OutputBase):
|
||||
REMOTE = RemoteLOCAL
|
||||
|
||||
def __init__(self,
|
||||
stage,
|
||||
@@ -29,9 +18,41 @@ class OutputLOCAL(DependencyLOCAL):
|
||||
remote=None,
|
||||
cache=True,
|
||||
metric=False):
|
||||
super(OutputLOCAL, self).__init__(stage, path, info, remote=remote)
|
||||
self.use_cache = cache
|
||||
self.metric = metric
|
||||
super(OutputLOCAL, self).__init__(stage,
|
||||
path,
|
||||
info,
|
||||
remote=remote,
|
||||
cache=cache,
|
||||
metric=metric)
|
||||
if remote:
|
||||
p = os.path.join(remote.prefix,
|
||||
urlparse(self.url).path.lstrip('/'))
|
||||
else:
|
||||
p = path
|
||||
|
||||
if not os.path.isabs(p):
|
||||
p = self.remote.to_ospath(p)
|
||||
p = os.path.join(stage.cwd, p)
|
||||
p = os.path.abspath(os.path.normpath(p))
|
||||
|
||||
self.path_info = {'scheme': 'local',
|
||||
'path': p}
|
||||
|
||||
def __str__(self):
|
||||
return self.rel_path
|
||||
|
||||
@property
|
||||
def is_local(self):
|
||||
return (urlparse(self.url).scheme != 'remote'
|
||||
and not os.path.isabs(self.url))
|
||||
|
||||
@property
|
||||
def sep(self):
|
||||
return os.sep
|
||||
|
||||
@property
|
||||
def rel_path(self):
|
||||
return os.path.relpath(self.path)
|
||||
|
||||
@property
|
||||
def md5(self):
|
||||
@@ -43,32 +64,16 @@ class OutputLOCAL(DependencyLOCAL):
|
||||
|
||||
def dumpd(self):
|
||||
ret = super(OutputLOCAL, self).dumpd()
|
||||
ret[self.PARAM_CACHE] = self.use_cache
|
||||
if self.is_local:
|
||||
path = self.remote.unixpath(os.path.relpath(self.path,
|
||||
self.stage.cwd))
|
||||
else:
|
||||
path = self.url
|
||||
|
||||
if isinstance(self.metric, dict):
|
||||
if self.PARAM_METRIC_XPATH in self.metric and \
|
||||
not self.metric[self.PARAM_METRIC_XPATH]:
|
||||
del self.metric[self.PARAM_METRIC_XPATH]
|
||||
|
||||
if self.metric:
|
||||
ret[self.PARAM_METRIC] = self.metric
|
||||
ret[self.PARAM_PATH] = path
|
||||
|
||||
return ret
|
||||
|
||||
def changed(self):
|
||||
if not self.use_cache:
|
||||
return super(OutputLOCAL, self).changed()
|
||||
|
||||
return self.project.cache.local.changed(self.path_info, self.info)
|
||||
|
||||
def checkout(self, force=False):
|
||||
if not self.use_cache:
|
||||
return
|
||||
|
||||
self.project.cache.local.checkout(self.path_info,
|
||||
self.info,
|
||||
force=force)
|
||||
|
||||
def _verify_metric(self):
|
||||
if not self.metric:
|
||||
return
|
||||
@@ -85,13 +90,6 @@ class OutputLOCAL(DependencyLOCAL):
|
||||
raise DvcException(msg.format(self.rel_path))
|
||||
|
||||
def save(self):
|
||||
if not self.use_cache:
|
||||
super(OutputLOCAL, self).save()
|
||||
self._verify_metric()
|
||||
msg = "Output '{}' doesn't use cache. Skipping saving."
|
||||
logger.info(msg.format(self.rel_path))
|
||||
return
|
||||
|
||||
if not os.path.exists(self.path):
|
||||
raise self.DoesNotExistError(self.rel_path)
|
||||
|
||||
@@ -103,6 +101,16 @@ class OutputLOCAL(DependencyLOCAL):
|
||||
msg = "file/directory '{}' is empty.".format(self.rel_path)
|
||||
logger.warning(msg)
|
||||
|
||||
if not self.use_cache:
|
||||
self.info = self.remote.save_info(self.path_info)
|
||||
self._verify_metric()
|
||||
if not self.IS_DEPENDENCY:
|
||||
msg = "Output '{}' doesn't use cache. Skipping saving."
|
||||
logger.info(msg.format(self.rel_path))
|
||||
return
|
||||
|
||||
assert not self.IS_DEPENDENCY
|
||||
|
||||
if not self.changed():
|
||||
msg = "Output '{}' didn't change. Skipping saving."
|
||||
logger.info(msg.format(self.rel_path))
|
||||
@@ -116,20 +124,3 @@ class OutputLOCAL(DependencyLOCAL):
|
||||
self.project.scm.ignore(self.path)
|
||||
|
||||
self.info = self.project.cache.local.save(self.path_info)
|
||||
|
||||
def remove(self, ignore_remove=False):
|
||||
self.remote.remove(self.path_info)
|
||||
if ignore_remove and self.use_cache and self.is_local:
|
||||
self.project.scm.ignore_remove(self.path)
|
||||
|
||||
def move(self, out):
|
||||
if self.use_cache and self.is_local:
|
||||
self.project.scm.ignore_remove(self.path)
|
||||
|
||||
self.remote.move(self.path_info, out.path_info)
|
||||
self.url = out.url
|
||||
self.path_info = out.path_info
|
||||
self.save()
|
||||
|
||||
if self.use_cache and self.is_local:
|
||||
self.project.scm.ignore(self.path)
|
||||
|
||||
+20
-43
@@ -1,10 +1,12 @@
|
||||
from dvc.dependency.s3 import DependencyS3
|
||||
from dvc.exceptions import DvcException
|
||||
import posixpath
|
||||
|
||||
from dvc.utils import urlparse
|
||||
from dvc.remote.s3 import RemoteS3
|
||||
from dvc.output.base import OutputBase
|
||||
|
||||
|
||||
class OutputS3(DependencyS3):
|
||||
PARAM_CACHE = 'cache'
|
||||
PARAM_METRIC = 'metric'
|
||||
class OutputS3(OutputBase):
|
||||
REMOTE = RemoteS3
|
||||
|
||||
def __init__(self,
|
||||
stage,
|
||||
@@ -13,41 +15,16 @@ class OutputS3(DependencyS3):
|
||||
remote=None,
|
||||
cache=True,
|
||||
metric=False):
|
||||
super(OutputS3, self).__init__(stage, path, info, remote=remote)
|
||||
self.use_cache = cache
|
||||
self.metric = metric
|
||||
if cache and self.project.cache.s3 is None:
|
||||
raise DvcException("no cache location setup for 's3' outputs.")
|
||||
|
||||
def dumpd(self):
|
||||
ret = super(OutputS3, self).dumpd()
|
||||
ret[self.PARAM_CACHE] = self.use_cache
|
||||
ret[self.PARAM_METRIC] = self.metric
|
||||
return ret
|
||||
|
||||
def changed(self):
|
||||
if super(OutputS3, self).changed():
|
||||
return True
|
||||
|
||||
if self.use_cache \
|
||||
and self.project.cache.s3.changed(self.path_info, self.info):
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def checkout(self, force=False):
|
||||
if not self.use_cache:
|
||||
return
|
||||
|
||||
self.project.cache.s3.checkout(self.path_info, self.info)
|
||||
|
||||
def save(self):
|
||||
super(OutputS3, self).save()
|
||||
|
||||
if not self.use_cache:
|
||||
return
|
||||
|
||||
self.info = self.project.cache.s3.save(self.path_info)
|
||||
|
||||
def remove(self, ignore_remove=False):
|
||||
self.remote.remove(self.path_info)
|
||||
super(OutputS3, self).__init__(stage,
|
||||
path,
|
||||
info=info,
|
||||
remote=remote,
|
||||
cache=cache,
|
||||
metric=metric)
|
||||
bucket = remote.bucket if remote else urlparse(path).netloc
|
||||
path = urlparse(path).path.lstrip('/')
|
||||
if remote:
|
||||
path = posixpath.join(remote.prefix, path)
|
||||
self.path_info = {'scheme': self.scheme,
|
||||
'bucket': bucket,
|
||||
'path': path}
|
||||
|
||||
+26
-41
@@ -1,10 +1,12 @@
|
||||
from dvc.dependency.ssh import DependencySSH
|
||||
from dvc.exceptions import DvcException
|
||||
import posixpath
|
||||
|
||||
from dvc.utils import urlparse
|
||||
from dvc.output.base import OutputBase
|
||||
from dvc.remote.ssh import RemoteSSH
|
||||
|
||||
|
||||
class OutputSSH(DependencySSH):
|
||||
PARAM_CACHE = 'cache'
|
||||
PARAM_METRIC = 'metric'
|
||||
class OutputSSH(OutputBase):
|
||||
REMOTE = RemoteSSH
|
||||
|
||||
def __init__(self,
|
||||
stage,
|
||||
@@ -13,41 +15,24 @@ class OutputSSH(DependencySSH):
|
||||
remote=None,
|
||||
cache=True,
|
||||
metric=False):
|
||||
super(OutputSSH, self).__init__(stage, path, info, remote=remote)
|
||||
self.use_cache = cache
|
||||
self.metric = metric
|
||||
if cache and self.project.cache.ssh is None:
|
||||
raise DvcException("no cache location setup for 'ssh' outputs.")
|
||||
super(OutputSSH, self).__init__(stage,
|
||||
path,
|
||||
info=info,
|
||||
remote=remote,
|
||||
cache=cache,
|
||||
metric=metric)
|
||||
host = remote.host if remote else self.group('host')
|
||||
port = remote.port if remote else RemoteSSH.DEFAULT_PORT
|
||||
user = remote.user if remote else self.group('user')
|
||||
|
||||
def dumpd(self):
|
||||
ret = super(OutputSSH, self).dumpd()
|
||||
ret[self.PARAM_CACHE] = self.use_cache
|
||||
ret[self.PARAM_METRIC] = self.metric
|
||||
return ret
|
||||
if remote:
|
||||
path = posixpath.join(remote.prefix,
|
||||
urlparse(path).path.lstrip('/'))
|
||||
else:
|
||||
path = self.match(self.url).group('path')
|
||||
|
||||
def changed(self):
|
||||
if super(OutputSSH, self).changed():
|
||||
return True
|
||||
|
||||
if self.use_cache \
|
||||
and self.project.cache.ssh.changed(self.path_info, self.info):
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def checkout(self, force=False):
|
||||
if not self.use_cache:
|
||||
return
|
||||
|
||||
self.project.cache.ssh.checkout(self.path_info, self.info)
|
||||
|
||||
def save(self):
|
||||
super(OutputSSH, self).save()
|
||||
|
||||
if not self.use_cache:
|
||||
return
|
||||
|
||||
self.info = self.project.cache.ssh.save(self.path_info)
|
||||
|
||||
def remove(self, ignore_remove=False):
|
||||
self.remote.remove(self.path_info)
|
||||
self.path_info = {'scheme': 'ssh',
|
||||
'host': host,
|
||||
'port': port,
|
||||
'user': user,
|
||||
'path': path}
|
||||
|
||||
+1
-5
@@ -7,12 +7,8 @@ try:
|
||||
except ImportError:
|
||||
BlockBlobService = None
|
||||
|
||||
try:
|
||||
from urlparse import urlparse
|
||||
except ImportError:
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import dvc.logger as logger
|
||||
from dvc.utils import urlparse
|
||||
from dvc.progress import progress
|
||||
from dvc.config import Config
|
||||
from dvc.remote.base import RemoteBase
|
||||
|
||||
+1
-5
@@ -5,12 +5,8 @@ try:
|
||||
except ImportError:
|
||||
storage = None
|
||||
|
||||
try:
|
||||
from urlparse import urlparse
|
||||
except ImportError:
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import dvc.logger as logger
|
||||
from dvc.utils import urlparse
|
||||
from dvc.remote.base import RemoteBase
|
||||
from dvc.config import Config
|
||||
from dvc.progress import progress
|
||||
|
||||
+1
-1
@@ -19,7 +19,7 @@ from concurrent.futures import ThreadPoolExecutor
|
||||
|
||||
|
||||
class RemoteLOCAL(RemoteBase):
|
||||
scheme = ''
|
||||
scheme = 'local'
|
||||
REGEX = r'^(?P<path>.*)$'
|
||||
PARAM_CHECKSUM = 'md5'
|
||||
PARAM_PATH = 'path'
|
||||
|
||||
+1
-5
@@ -6,12 +6,8 @@ try:
|
||||
except ImportError:
|
||||
boto3 = None
|
||||
|
||||
try:
|
||||
from urlparse import urlparse
|
||||
except ImportError:
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import dvc.logger as logger
|
||||
from dvc.utils import urlparse
|
||||
from dvc.progress import progress
|
||||
from dvc.config import Config
|
||||
from dvc.remote.base import RemoteBase
|
||||
|
||||
@@ -9,6 +9,11 @@ import json
|
||||
import shutil
|
||||
import hashlib
|
||||
|
||||
try:
|
||||
from urlparse import urlparse, urljoin # noqa: F401
|
||||
except ImportError:
|
||||
from urllib.parse import urlparse, urljoin # noqa: F401
|
||||
|
||||
import dvc.logger as logger
|
||||
from dvc.progress import progress
|
||||
from dvc.istextfile import istextfile
|
||||
|
||||
@@ -12,14 +12,14 @@ class TestOutScheme(TestDvc):
|
||||
'gs://bucket/path': 'gs',
|
||||
'ssh://example.com:/dir/path': 'ssh',
|
||||
'hdfs://example.com/dir/path': 'hdfs',
|
||||
'path/to/file': '',
|
||||
'path\\to\\file': '',
|
||||
'file': '',
|
||||
'./file': '',
|
||||
'.\\file': '',
|
||||
'../file': '',
|
||||
'..\\file': '',
|
||||
'unknown://path': '',
|
||||
'path/to/file': 'local',
|
||||
'path\\to\\file': 'local',
|
||||
'file': 'local',
|
||||
'./file': 'local',
|
||||
'.\\file': 'local',
|
||||
'../file': 'local',
|
||||
'..\\file': 'local',
|
||||
'unknown://path': 'local',
|
||||
}
|
||||
|
||||
def _get(self, path):
|
||||
|
||||
Reference in New Issue
Block a user