From 7cd24e0c1fda50e9cb79cd74f5433122f9751ee8 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sat, 5 Jan 2019 22:54:39 +0200 Subject: [PATCH 1/3] 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 --- dvc/command/imp.py | 6 +- dvc/dependency/__init__.py | 26 ++----- dvc/dependency/base.py | 72 +---------------- dvc/dependency/gs.py | 26 +------ dvc/dependency/hdfs.py | 32 +------- dvc/dependency/http.py | 30 +++----- dvc/dependency/local.py | 90 +--------------------- dvc/dependency/s3.py | 34 +-------- dvc/dependency/ssh.py | 41 +--------- dvc/output/__init__.py | 27 ++++--- dvc/output/base.py | 153 +++++++++++++++++++++++++++++++++++++ dvc/output/gs.py | 55 +------------ dvc/output/hdfs.py | 62 +++++---------- dvc/output/local.py | 123 ++++++++++++++--------------- dvc/output/s3.py | 63 +++++---------- dvc/output/ssh.py | 67 +++++++--------- dvc/remote/azure.py | 6 +- dvc/remote/gs.py | 6 +- dvc/remote/local.py | 2 +- dvc/remote/s3.py | 6 +- dvc/utils.py | 5 ++ tests/test_output.py | 16 ++-- 22 files changed, 351 insertions(+), 597 deletions(-) diff --git a/dvc/command/imp.py b/dvc/command/imp.py index 7f9a2dfa6..0b42922f7 100644 --- a/dvc/command/imp.py +++ b/dvc/command/imp.py @@ -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 diff --git a/dvc/dependency/__init__.py b/dvc/dependency/__init__.py index 52c16bf30..800316400 100644 --- a/dvc/dependency/__init__.py +++ b/dvc/dependency/__init__.py @@ -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 diff --git a/dvc/dependency/base.py b/dvc/dependency/base.py index a1224ac7a..4c860cee3 100644 --- a/dvc/dependency/base.py +++ b/dvc/dependency/base.py @@ -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 diff --git a/dvc/dependency/gs.py b/dvc/dependency/gs.py index 0bcc14d70..4f26b05ef 100644 --- a/dvc/dependency/gs.py +++ b/dvc/dependency/gs.py @@ -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 diff --git a/dvc/dependency/hdfs.py b/dvc/dependency/hdfs.py index dc4a7c41a..3f5659334 100644 --- a/dvc/dependency/hdfs.py +++ b/dvc/dependency/hdfs.py @@ -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 diff --git a/dvc/dependency/http.py b/dvc/dependency/http.py index df2951ae9..1a663b2c7 100644 --- a/dvc/dependency/http.py +++ b/dvc/dependency/http.py @@ -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) diff --git a/dvc/dependency/local.py b/dvc/dependency/local.py index 4ba7e9bdf..02c77176a 100644 --- a/dvc/dependency/local.py +++ b/dvc/dependency/local.py @@ -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.*)$' - - 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 diff --git a/dvc/dependency/s3.py b/dvc/dependency/s3.py index 62b9330bc..5a40e647e 100644 --- a/dvc/dependency/s3.py +++ b/dvc/dependency/s3.py @@ -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 diff --git a/dvc/dependency/ssh.py b/dvc/dependency/ssh.py index 2f25a9707..1e62ab5e1 100644 --- a/dvc/dependency/ssh.py +++ b/dvc/dependency/ssh.py @@ -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 diff --git a/dvc/output/__init__.py b/dvc/output/__init__.py index 20e7c92f2..6def97fca 100644 --- a/dvc/output/__init__.py +++ b/dvc/output/__init__.py @@ -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 diff --git a/dvc/output/base.py b/dvc/output/base.py index 2153a8d95..28d3c8791 100644 --- a/dvc/output/base.py +++ b/dvc/output/base.py @@ -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) diff --git a/dvc/output/gs.py b/dvc/output/gs.py index 2e483377d..74bb41218 100644 --- a/dvc/output/gs.py +++ b/dvc/output/gs.py @@ -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 diff --git a/dvc/output/hdfs.py b/dvc/output/hdfs.py index e3e6bce9d..725a181c0 100644 --- a/dvc/output/hdfs.py +++ b/dvc/output/hdfs.py @@ -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} diff --git a/dvc/output/local.py b/dvc/output/local.py index bd9ff504d..cab178871 100644 --- a/dvc/output/local.py +++ b/dvc/output/local.py @@ -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) diff --git a/dvc/output/s3.py b/dvc/output/s3.py index 25c3be04d..41cc31ed3 100644 --- a/dvc/output/s3.py +++ b/dvc/output/s3.py @@ -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} diff --git a/dvc/output/ssh.py b/dvc/output/ssh.py index 90af3b964..53d10daee 100644 --- a/dvc/output/ssh.py +++ b/dvc/output/ssh.py @@ -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} diff --git a/dvc/remote/azure.py b/dvc/remote/azure.py index a13c8ae62..01e71e4cc 100644 --- a/dvc/remote/azure.py +++ b/dvc/remote/azure.py @@ -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 diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index 8d808b6d8..65b9e71d6 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -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 diff --git a/dvc/remote/local.py b/dvc/remote/local.py index df6c482da..86acc79a0 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -19,7 +19,7 @@ from concurrent.futures import ThreadPoolExecutor class RemoteLOCAL(RemoteBase): - scheme = '' + scheme = 'local' REGEX = r'^(?P.*)$' PARAM_CHECKSUM = 'md5' PARAM_PATH = 'path' diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 621895894..9bff1ebc6 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -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 diff --git a/dvc/utils.py b/dvc/utils.py index 120fff3ef..17884237e 100644 --- a/dvc/utils.py +++ b/dvc/utils.py @@ -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 diff --git a/tests/test_output.py b/tests/test_output.py index a93068aaa..e2bb7c9cc 100644 --- a/tests/test_output.py +++ b/tests/test_output.py @@ -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): From 68d965f4fba00d22a1752476c2908e342a4ec21a Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 7 Jan 2019 18:20:56 +0200 Subject: [PATCH 2/3] RFC: utils: create a separate module Signed-off-by: Ruslan Kuprieiev --- dvc/command/imp.py | 2 +- dvc/dependency/__init__.py | 2 +- dvc/dependency/http.py | 2 +- dvc/output/__init__.py | 2 +- dvc/output/hdfs.py | 2 +- dvc/output/local.py | 2 +- dvc/output/s3.py | 2 +- dvc/output/ssh.py | 2 +- dvc/remote/azure.py | 2 +- dvc/remote/gs.py | 2 +- dvc/remote/s3.py | 2 +- dvc/{utils.py => utils/__init__.py} | 5 ----- dvc/utils/compat.py | 6 ++++++ tests/test_repro.py | 6 +----- 14 files changed, 18 insertions(+), 21 deletions(-) rename dvc/{utils.py => utils/__init__.py} (96%) create mode 100644 dvc/utils/compat.py diff --git a/dvc/command/imp.py b/dvc/command/imp.py index 0b42922f7..77390b620 100644 --- a/dvc/command/imp.py +++ b/dvc/command/imp.py @@ -1,7 +1,7 @@ import os import dvc.logger as logger -from dvc.utils import urlparse +from dvc.utils.compat import urlparse from dvc.exceptions import DvcException from dvc.command.base import CmdBase diff --git a/dvc/dependency/__init__.py b/dvc/dependency/__init__.py index 800316400..929e7e8a0 100644 --- a/dvc/dependency/__init__.py +++ b/dvc/dependency/__init__.py @@ -1,7 +1,7 @@ import schema from dvc.config import Config -from dvc.utils import urlparse +from dvc.utils.compat import urlparse import dvc.output as output from dvc.output.base import OutputBase diff --git a/dvc/dependency/http.py b/dvc/dependency/http.py index 1a663b2c7..562e690c0 100644 --- a/dvc/dependency/http.py +++ b/dvc/dependency/http.py @@ -1,4 +1,4 @@ -from dvc.utils import urlparse, urljoin +from dvc.utils.compat import urlparse, urljoin from dvc.output.base import OutputBase from dvc.remote.http import RemoteHTTP from dvc.dependency.base import DependencyBase diff --git a/dvc/output/__init__.py b/dvc/output/__init__.py index 6def97fca..0bf820ee8 100644 --- a/dvc/output/__init__.py +++ b/dvc/output/__init__.py @@ -1,7 +1,7 @@ import schema from dvc.config import Config -from dvc.utils import urlparse +from dvc.utils.compat import urlparse from dvc.output.base import OutputBase from dvc.output.s3 import OutputS3 diff --git a/dvc/output/hdfs.py b/dvc/output/hdfs.py index 725a181c0..4759f4e82 100644 --- a/dvc/output/hdfs.py +++ b/dvc/output/hdfs.py @@ -1,6 +1,6 @@ import posixpath -from dvc.utils import urlparse +from dvc.utils.compat import urlparse from dvc.output.base import OutputBase from dvc.remote.hdfs import RemoteHDFS diff --git a/dvc/output/local.py b/dvc/output/local.py index cab178871..4e1f8d3ff 100644 --- a/dvc/output/local.py +++ b/dvc/output/local.py @@ -1,7 +1,7 @@ import os import dvc.logger as logger -from dvc.utils import urlparse +from dvc.utils.compat import urlparse from dvc.istextfile import istextfile from dvc.exceptions import DvcException from dvc.remote.local import RemoteLOCAL diff --git a/dvc/output/s3.py b/dvc/output/s3.py index 41cc31ed3..7c46e11e3 100644 --- a/dvc/output/s3.py +++ b/dvc/output/s3.py @@ -1,7 +1,7 @@ import posixpath -from dvc.utils import urlparse from dvc.remote.s3 import RemoteS3 +from dvc.utils.compat import urlparse from dvc.output.base import OutputBase diff --git a/dvc/output/ssh.py b/dvc/output/ssh.py index 53d10daee..f26a48bed 100644 --- a/dvc/output/ssh.py +++ b/dvc/output/ssh.py @@ -1,6 +1,6 @@ import posixpath -from dvc.utils import urlparse +from dvc.utils.compat import urlparse from dvc.output.base import OutputBase from dvc.remote.ssh import RemoteSSH diff --git a/dvc/remote/azure.py b/dvc/remote/azure.py index 01e71e4cc..abb9d5c13 100644 --- a/dvc/remote/azure.py +++ b/dvc/remote/azure.py @@ -8,7 +8,7 @@ except ImportError: BlockBlobService = None import dvc.logger as logger -from dvc.utils import urlparse +from dvc.utils.compat import urlparse from dvc.progress import progress from dvc.config import Config from dvc.remote.base import RemoteBase diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index 65b9e71d6..8530f36b2 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -6,7 +6,7 @@ except ImportError: storage = None import dvc.logger as logger -from dvc.utils import urlparse +from dvc.utils.compat import urlparse from dvc.remote.base import RemoteBase from dvc.config import Config from dvc.progress import progress diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 9bff1ebc6..420985520 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -7,7 +7,7 @@ except ImportError: boto3 = None import dvc.logger as logger -from dvc.utils import urlparse +from dvc.utils.compat import urlparse from dvc.progress import progress from dvc.config import Config from dvc.remote.base import RemoteBase diff --git a/dvc/utils.py b/dvc/utils/__init__.py similarity index 96% rename from dvc/utils.py rename to dvc/utils/__init__.py index 17884237e..120fff3ef 100644 --- a/dvc/utils.py +++ b/dvc/utils/__init__.py @@ -9,11 +9,6 @@ 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 diff --git a/dvc/utils/compat.py b/dvc/utils/compat.py new file mode 100644 index 000000000..c075ae306 --- /dev/null +++ b/dvc/utils/compat.py @@ -0,0 +1,6 @@ +"""Handle impor compatibility between Python 2 and Python 3""" + +try: + from urlparse import urlparse, urljoin # noqa: F401 +except ImportError: + from urllib.parse import urlparse, urljoin # noqa: F401 diff --git a/tests/test_repro.py b/tests/test_repro.py index 6d91e060d..a943657e2 100644 --- a/tests/test_repro.py +++ b/tests/test_repro.py @@ -6,11 +6,6 @@ import getpass import posixpath from subprocess import Popen, PIPE -try: - from urlparse import urljoin -except ImportError: - from urllib.parse import urljoin - import boto3 import uuid from google.cloud import storage as gc @@ -18,6 +13,7 @@ from google.cloud import storage as gc from dvc.main import main from dvc.project import Project, ReproductionError from dvc.utils import file_md5 +from dvc.utils.compat import urljoin from dvc.remote.local import RemoteLOCAL from dvc.stage import Stage, StageFileDoesNotExistError from dvc.system import System From ce009c894f7705b6305e960fad5c4981c4535533 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 7 Jan 2019 18:40:45 +0200 Subject: [PATCH 3/3] RFC: deps/outs: add more comments Signed-off-by: Ruslan Kuprieiev --- dvc/dependency/__init__.py | 5 +++++ dvc/output/__init__.py | 11 ++++++++++- 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/dvc/dependency/__init__.py b/dvc/dependency/__init__.py index 929e7e8a0..8139fb734 100644 --- a/dvc/dependency/__init__.py +++ b/dvc/dependency/__init__.py @@ -33,6 +33,11 @@ DEP_MAP = { 'https': DependencyHTTP, } + +# NOTE: schema for dependencies is basically the same as for outputs, but +# without output-specific entries like 'cache' (whether or not output is +# cached, see -o and -O flags for `dvc run`) and 'metric' (whether or not +# output is a metric file and how to parse it, see `-M` flag for `dvc run`). SCHEMA = output.SCHEMA.copy() del SCHEMA[schema.Optional(OutputBase.PARAM_CACHE)] del SCHEMA[schema.Optional(OutputBase.PARAM_METRIC)] diff --git a/dvc/output/__init__.py b/dvc/output/__init__.py index 0bf820ee8..e17c21831 100644 --- a/dvc/output/__init__.py +++ b/dvc/output/__init__.py @@ -29,12 +29,21 @@ OUTS_MAP = {'hdfs': OutputHDFS, 'ssh': OutputSSH, 'local': OutputLOCAL} -# We are skipping RemoteHTTP.PARAM_CHECKSUM because is the same as RemoteS3 SCHEMA = { OutputBase.PARAM_PATH: str, + + # NOTE: currently there are only 3 possible checksum names: + # + # 1) md5 (LOCAL, SSH, GS); + # 2) etag (S3); + # 3) checksum (HDFS); + # + # so when a few types of outputs share the same name, we only need + # specify it once. 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, }