diff --git a/dvc/command/diff.py b/dvc/command/diff.py index 2635e9c72..a1d37ae8f 100644 --- a/dvc/command/diff.py +++ b/dvc/command/diff.py @@ -5,10 +5,10 @@ import argparse import humanize import inflect import logging +from funcy import compact from dvc.exceptions import DvcException from dvc.command.base import CmdBase, append_doc_link -from dvc.utils.collections import compact import dvc.repo.diff as diff diff --git a/dvc/command/version.py b/dvc/command/version.py index 34be52980..77424018d 100644 --- a/dvc/command/version.py +++ b/dvc/command/version.py @@ -7,7 +7,7 @@ import argparse import logging import uuid -from dvc.utils.compat import Path +from dvc.utils.compat import pathlib from dvc.repo import Repo from dvc.command.base import CmdBaseNoRepo, append_doc_link from dvc.version import __version__ @@ -58,10 +58,10 @@ class CmdVersion(CmdBaseNoRepo): @staticmethod def get_fs_type(path): partition = { - Path(part.mountpoint): (part.fstype, part.device) + pathlib.Path(part.mountpoint): (part.fstype, part.device) for part in psutil.disk_partitions() } - for parent in list(Path(path).parents): + for parent in pathlib.Path(path).parents: if parent in partition: return partition[parent] return ("unkown", "none") diff --git a/dvc/config.py b/dvc/config.py index b6f68759f..a86777d4d 100644 --- a/dvc/config.py +++ b/dvc/config.py @@ -255,6 +255,7 @@ class Config(object): # pylint: disable=too-many-instance-attributes SECTION_REMOTE_TIMEOUT = "timeout" SECTION_REMOTE_PASSWORD = "password" SECTION_REMOTE_ASK_PASSWORD = "ask_password" + SECTION_REMOTE_NO_TRAVERSE = "no_traverse" SECTION_REMOTE_SCHEMA = { SECTION_REMOTE_URL: str, Optional(SECTION_AWS_REGION): str, @@ -278,6 +279,7 @@ class Config(object): # pylint: disable=too-many-instance-attributes Optional(SECTION_OSS_ACCESS_KEY_SECRET): str, Optional(SECTION_OSS_ENDPOINT): str, Optional(PRIVATE_CWD): str, + Optional(SECTION_REMOTE_NO_TRAVERSE, default=True): BOOL_SCHEMA, } SECTION_STATE = "state" diff --git a/dvc/dependency/http.py b/dvc/dependency/http.py index 24ddfe497..47133adc6 100644 --- a/dvc/dependency/http.py +++ b/dvc/dependency/http.py @@ -1,7 +1,5 @@ from __future__ import unicode_literals -from dvc.path import Path -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 @@ -9,12 +7,3 @@ from dvc.dependency.base import DependencyBase class DependencyHTTP(DependencyBase, OutputBase): REMOTE = RemoteHTTP - - def __init__(self, stage, path, info=None, remote=None): - super(DependencyHTTP, self).__init__( - stage, path, info=info, remote=remote - ) - if path.startswith("remote"): - path = urljoin(self.remote.cache_dir, urlparse(path).path) - - self.path_info = Path(self.scheme, url=self.url, path=path) diff --git a/dvc/exceptions.py b/dvc/exceptions.py index 70e1aaa18..98f4a8c56 100644 --- a/dvc/exceptions.py +++ b/dvc/exceptions.py @@ -98,9 +98,7 @@ class CircularDependencyError(DvcException): """ def __init__(self, dependency): - assert isinstance(dependency, str) or isinstance( - dependency, builtin_str - ) + assert isinstance(dependency, (str, builtin_str)) msg = ( "file/directory '{}' is specified as an output and as a " @@ -118,7 +116,7 @@ class ArgumentDuplicationError(DvcException): """ def __init__(self, path): - assert isinstance(path, str) or isinstance(path, builtin_str) + assert isinstance(path, (str, builtin_str)) msg = "file '{}' is specified more than once." super(ArgumentDuplicationError, self).__init__(msg.format(path)) diff --git a/dvc/output/base.py b/dvc/output/base.py index 784c12e99..f9360a78c 100644 --- a/dvc/output/base.py +++ b/dvc/output/base.py @@ -1,13 +1,12 @@ from __future__ import unicode_literals -import re import logging from copy import copy from schema import Or, Optional from dvc.exceptions import DvcException -from dvc.utils.compat import str +from dvc.utils.compat import str, urlparse from dvc.remote.base import RemoteBASE @@ -58,6 +57,8 @@ class OutputBase(object): DoesNotExistError = OutputDoesNotExistError IsNotFileOrDirError = OutputIsNotFileOrDirError + sep = "/" + def __init__( self, stage, @@ -69,9 +70,19 @@ class OutputBase(object): persist=False, tags=None, ): + # This output (and dependency) objects have too many paths/urls + # here is a list and comments: + # + # .def_path - path from definition in stage file + # .path_info - PathInfo/URLInfo structured resolved path + # .fspath - local only, resolved + # .__str__ - for presentation purposes, def_path/relpath + # + # By resolved path, which contains actual location, + # should be absolute and don't contain remote:// refs. self.stage = stage self.repo = stage.repo - self.url = path + self.def_path = path self.info = info self.remote = remote or self.REMOTE(self.repo, {}) self.use_cache = False if self.IS_DEPENDENCY else cache @@ -86,13 +97,26 @@ class OutputBase(object): ) ) + self.path_info = self._parse_path(remote, path) + + def _parse_path(self, remote, path): + if remote: + parsed = urlparse(path) + return remote.path_info / parsed.path.lstrip("/") + else: + return self.REMOTE.path_cls(path) + def __repr__(self): - return "{class_name}: '{url}'".format( - class_name=type(self).__name__, url=(self.url or "No url") + return "{class_name}: '{def_path}'".format( + class_name=type(self).__name__, def_path=self.def_path ) def __str__(self): - return self.url + return self.def_path + + @property + def scheme(self): + return self.REMOTE.scheme @property def is_in_repo(self): @@ -111,35 +135,13 @@ class OutputBase(object): "change repo is not supported for {}".format(self.scheme) ) - @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 + return cls.REMOTE.supported(url) @property def cache_path(self): - return self.cache.checksum_to_path(self.checksum) - - @property - def sep(self): - return "/" + return self.cache.checksum_to_path_info(self.checksum).url @property def checksum(self): @@ -228,11 +230,11 @@ class OutputBase(object): return if self.is_in_repo: - if self.repo.scm.is_tracked(self.path): + if self.repo.scm.is_tracked(self.fspath): raise OutputAlreadyTrackedError(self) if self.use_cache: - self.repo.scm.ignore(self.path) + self.repo.scm.ignore(self.fspath) self.info = self.remote.save_info(self.path_info) @@ -242,7 +244,7 @@ class OutputBase(object): def dumpd(self): ret = copy(self.info) - ret[self.PARAM_PATH] = self.url + ret[self.PARAM_PATH] = self.def_path if self.IS_DEPENDENCY: return ret @@ -294,20 +296,20 @@ class OutputBase(object): return if ignore_remove and self.use_cache and self.is_in_repo: - self.repo.scm.ignore_remove(self.path) + self.repo.scm.ignore_remove(self.fspath) def move(self, out): if self.scheme == "local" and self.use_cache and self.is_in_repo: - self.repo.scm.ignore_remove(self.path) + self.repo.scm.ignore_remove(self.fspath) self.remote.move(self.path_info, out.path_info) - self.url = out.url - self.path_info = copy(out.path_info) + self.def_path = out.def_path + self.path_info = out.path_info self.save() self.commit() if self.scheme == "local" and self.use_cache and self.is_in_repo: - self.repo.scm.ignore(self.path) + self.repo.scm.ignore(self.fspath) def get_files_number(self): if not self.use_cache or not self.checksum: diff --git a/dvc/output/hdfs.py b/dvc/output/hdfs.py index 94225c90b..e26f5161b 100644 --- a/dvc/output/hdfs.py +++ b/dvc/output/hdfs.py @@ -1,38 +1,8 @@ from __future__ import unicode_literals -import posixpath - -from dvc.path.hdfs import PathHDFS -from dvc.utils.compat import urlparse from dvc.output.base import OutputBase from dvc.remote.hdfs import RemoteHDFS class OutputHDFS(OutputBase): REMOTE = RemoteHDFS - - def __init__( - self, - stage, - path, - info=None, - remote=None, - cache=True, - metric=False, - persist=False, - tags=None, - ): - super(OutputHDFS, self).__init__( - stage, - path, - info=info, - remote=remote, - cache=cache, - metric=metric, - persist=persist, - tags=tags, - ) - if remote: - path = posixpath.join(remote.url, urlparse(path).path.lstrip("/")) - user = remote.user if remote else self.group("user") - self.path_info = PathHDFS(user=user, url=self.url, path=path) diff --git a/dvc/output/local.py b/dvc/output/local.py index 1064009ca..13ae650f2 100644 --- a/dvc/output/local.py +++ b/dvc/output/local.py @@ -3,8 +3,7 @@ from __future__ import unicode_literals import os import logging -from dvc.path.local import PathLOCAL -from dvc.utils.compat import urlparse +from dvc.utils.compat import urlparse, fspath_py35, str from dvc.istextfile import istextfile from dvc.exceptions import DvcException from dvc.remote.local import RemoteLOCAL @@ -16,82 +15,58 @@ logger = logging.getLogger(__name__) class OutputLOCAL(OutputBase): REMOTE = RemoteLOCAL + sep = os.sep - def __init__( - self, - stage, - path, - info=None, - remote=None, - cache=True, - metric=False, - persist=False, - tags=None, - ): - super(OutputLOCAL, self).__init__( - stage, - path, - info, - remote=remote, - cache=cache, - metric=metric, - persist=persist, - tags=tags, - ) - if remote: - p = os.path.join( - remote.prefix, urlparse(self.url).path.lstrip("/") - ) + def _parse_path(self, remote, path): + parsed = urlparse(path) + if parsed.scheme == "remote": + p = remote.path_info / parsed.path.lstrip("/") else: - p = path + # NOTE: we can path either from command line or .dvc file, + # so we should expect both posix and windows style paths. + # PathInfo accepts both, i.e. / works everywhere, \ only on win. + # + # FIXME: if we have Windows path containig / or posix one with \ + # then we have #2059 bug and can't really handle that. + p = self.REMOTE.path_cls(path) + if not p.is_absolute(): + p = self.stage.wdir / p - if not os.path.isabs(p): - p = self.remote.to_ospath(p) - p = os.path.join(stage.wdir, p) - p = os.path.abspath(os.path.normpath(p)) - - self.path_info = PathLOCAL(url=self.url, path=p) + abs_p = os.path.abspath(os.path.normpath(fspath_py35(p))) + return self.REMOTE.path_cls(abs_p) def __str__(self): - return self.rel_path + return str(self.path_info) + + @property + def fspath(self): + return self.path_info.fspath @property def is_in_repo(self): - return urlparse(self.url).scheme != "remote" and not os.path.isabs( - self.url - ) + def_scheme = urlparse(self.def_path).scheme + return def_scheme != "remote" and not os.path.isabs(self.def_path) def assign_to_stage_file(self, stage): from dvc.repo import Repo - fullpath = os.path.abspath(stage.wdir) - self.path_info.path = os.path.join(fullpath, self.stage_path) + # Make path info the same but relative to passed stage + rel_path = self.path_info.relpath(self.stage.wdir) + self.path_info = ( + self.REMOTE.path_cls(os.path.abspath(stage.wdir)) / rel_path + ) - self.repo = Repo(self.path) + self.repo = Repo(self.fspath) self.stage = stage return self - @property - def sep(self): - return os.sep - - @property - def rel_path(self): - return os.path.relpath(self.path) - - @property - def stage_path(self): - return os.path.relpath(self.path, self.stage.wdir) - def dumpd(self): ret = super(OutputLOCAL, self).dumpd() if self.is_in_repo: - path = self.remote.to_posixpath( - os.path.relpath(self.path, self.stage.wdir) - ) + path = self.path_info.relpath(self.stage.wdir).as_posix() else: - path = self.url + path = self.def_path ret[self.PARAM_PATH] = path @@ -101,13 +76,14 @@ class OutputLOCAL(OutputBase): if not self.metric: return - if not os.path.exists(self.path): + path = fspath_py35(self.path_info) + if not os.path.exists(path): return - if os.path.isdir(self.path): + if os.path.isdir(path): msg = "directory '{}' cannot be used as metrics." - raise DvcException(msg.format(self.rel_path)) + raise DvcException(msg.format(self.path_info)) - if not istextfile(self.path): + if not istextfile(path): msg = "binary file '{}' cannot be used as metrics." - raise DvcException(msg.format(self.rel_path)) + raise DvcException(msg.format(self.path_info)) diff --git a/dvc/output/s3.py b/dvc/output/s3.py index 4c5cc4724..d9cd116b4 100644 --- a/dvc/output/s3.py +++ b/dvc/output/s3.py @@ -1,42 +1,8 @@ from __future__ import unicode_literals -import posixpath - -from dvc.path import Path from dvc.remote.s3 import RemoteS3 -from dvc.utils.compat import urlparse from dvc.output.base import OutputBase class OutputS3(OutputBase): REMOTE = RemoteS3 - - def __init__( - self, - stage, - path, - info=None, - remote=None, - cache=True, - metric=False, - persist=False, - tags=None, - ): - super(OutputS3, self).__init__( - stage, - path, - info=info, - remote=remote, - cache=cache, - metric=metric, - persist=persist, - tags=tags, - ) - 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 = Path( - self.scheme, bucket=bucket, path=path, url=self.url - ) diff --git a/dvc/output/ssh.py b/dvc/output/ssh.py index f461175fd..5b4526a20 100644 --- a/dvc/output/ssh.py +++ b/dvc/output/ssh.py @@ -1,54 +1,8 @@ from __future__ import unicode_literals -import getpass -import posixpath - -from dvc.path.ssh import PathSSH -from dvc.utils.compat import urlparse from dvc.output.base import OutputBase from dvc.remote.ssh import RemoteSSH class OutputSSH(OutputBase): REMOTE = RemoteSSH - - def __init__( - self, - stage, - path, - info=None, - remote=None, - cache=True, - metric=False, - persist=False, - tags=None, - ): - super(OutputSSH, self).__init__( - stage, - path, - info=info, - remote=remote, - cache=cache, - metric=metric, - persist=persist, - tags=tags, - ) - parsed = urlparse(path) - host = remote.host if remote else parsed.hostname - port = ( - remote.port if remote else (parsed.port or RemoteSSH.DEFAULT_PORT) - ) - user = ( - remote.user if remote else (parsed.username or getpass.getuser()) - ) - - if remote: - path = posixpath.join( - remote.prefix, urlparse(path).path.lstrip("/") - ) - else: - path = parsed.path - - self.path_info = PathSSH( - host=host, user=user, port=port, url=self.url, path=path - ) diff --git a/dvc/path/__init__.py b/dvc/path/__init__.py deleted file mode 100644 index 0af623dee..000000000 --- a/dvc/path/__init__.py +++ /dev/null @@ -1,29 +0,0 @@ -from dvc.scheme import Schemes - -from dvc.path.azure import PathAZURE -from dvc.path.gs import PathGS -from dvc.path.hdfs import PathHDFS -from dvc.path.http import PathHTTP -from dvc.path.https import PathHTTPS -from dvc.path.local import PathLOCAL -from dvc.path.oss import PathOSS -from dvc.path.s3 import PathS3 -from dvc.path.ssh import PathSSH - - -PATH_MAP = { - Schemes.SSH: PathSSH, - Schemes.HDFS: PathHDFS, - Schemes.S3: PathS3, - Schemes.AZURE: PathAZURE, - Schemes.HTTP: PathHTTP, - Schemes.HTTPS: PathHTTPS, - Schemes.GS: PathGS, - Schemes.LOCAL: PathLOCAL, - Schemes.OSS: PathOSS, -} - - -def Path(scheme, *args, **kwargs): - cls = PATH_MAP[scheme] - return cls(*args, **kwargs) diff --git a/dvc/path/azure.py b/dvc/path/azure.py deleted file mode 100644 index 019cf0af4..000000000 --- a/dvc/path/azure.py +++ /dev/null @@ -1,6 +0,0 @@ -from dvc.scheme import Schemes -from .base import PathCloudBASE - - -class PathAZURE(PathCloudBASE): - scheme = Schemes.AZURE diff --git a/dvc/path/base.py b/dvc/path/base.py deleted file mode 100644 index 616df4041..000000000 --- a/dvc/path/base.py +++ /dev/null @@ -1,23 +0,0 @@ -from dvc.utils.compat import urlunsplit - - -class PathBASE(object): - scheme = None - - def __init__(self, url=None, path=None): - self.url = url - self.path = path - - def __str__(self): - return self.url - - -class PathCloudBASE(PathBASE): - def __init__(self, bucket, url=None, path=None): - super(PathCloudBASE, self).__init__(url, path) - self.bucket = bucket - - def __str__(self): - if not self.url: - return urlunsplit((self.scheme, self.bucket, self.path, "", "")) - return self.url diff --git a/dvc/path/gs.py b/dvc/path/gs.py deleted file mode 100644 index f2befca6c..000000000 --- a/dvc/path/gs.py +++ /dev/null @@ -1,6 +0,0 @@ -from dvc.scheme import Schemes -from .base import PathCloudBASE - - -class PathGS(PathCloudBASE): - scheme = Schemes.GS diff --git a/dvc/path/hdfs.py b/dvc/path/hdfs.py deleted file mode 100644 index 3e9e8f63b..000000000 --- a/dvc/path/hdfs.py +++ /dev/null @@ -1,13 +0,0 @@ -from dvc.scheme import Schemes -from .base import PathBASE - - -class PathHDFS(PathBASE): - scheme = Schemes.HDFS - - def __init__(self, user, url=None, path=None): - super(PathHDFS, self).__init__(url, path) - self.user = user - - def __str__(self): - return self.path diff --git a/dvc/path/http.py b/dvc/path/http.py deleted file mode 100644 index 579db8198..000000000 --- a/dvc/path/http.py +++ /dev/null @@ -1,13 +0,0 @@ -from dvc.scheme import Schemes -from dvc.utils.compat import urlunsplit - -from .base import PathBASE - - -class PathHTTP(PathBASE): - scheme = Schemes.HTTP - - def __str__(self): - if not self.url: - return urlunsplit((self.scheme, self.path, "", "", "")) - return self.url diff --git a/dvc/path/https.py b/dvc/path/https.py deleted file mode 100644 index b631045eb..000000000 --- a/dvc/path/https.py +++ /dev/null @@ -1,6 +0,0 @@ -from dvc.scheme import Schemes -from .http import PathHTTP - - -class PathHTTPS(PathHTTP): - scheme = Schemes.HTTPS diff --git a/dvc/path/local.py b/dvc/path/local.py deleted file mode 100644 index 4129be1df..000000000 --- a/dvc/path/local.py +++ /dev/null @@ -1,17 +0,0 @@ -import os - -from dvc.scheme import Schemes -from .base import PathBASE - - -class PathLOCAL(PathBASE): - scheme = Schemes.LOCAL - - def __str__(self): - if os.name == "nt" and not os.path.commonprefix( - [os.getcwd(), self.path] - ): - # In case of windows, when cache is on different drive than - # workspace, we will get ValueError when trying to get relpath - return self.path - return os.path.relpath(self.path) diff --git a/dvc/path/oss.py b/dvc/path/oss.py deleted file mode 100644 index 67cb3dd19..000000000 --- a/dvc/path/oss.py +++ /dev/null @@ -1,6 +0,0 @@ -from dvc.scheme import Schemes -from .base import PathCloudBASE - - -class PathOSS(PathCloudBASE): - scheme = Schemes.OSS diff --git a/dvc/path/s3.py b/dvc/path/s3.py deleted file mode 100644 index df46e87e3..000000000 --- a/dvc/path/s3.py +++ /dev/null @@ -1,6 +0,0 @@ -from dvc.scheme import Schemes -from .base import PathCloudBASE - - -class PathS3(PathCloudBASE): - scheme = Schemes.S3 diff --git a/dvc/path/ssh.py b/dvc/path/ssh.py deleted file mode 100644 index 13f0b8508..000000000 --- a/dvc/path/ssh.py +++ /dev/null @@ -1,26 +0,0 @@ -from dvc.utils.compat import urlunsplit -from dvc.scheme import Schemes -from .base import PathBASE - - -class PathSSH(PathBASE): - scheme = Schemes.SSH - - def __init__(self, host, user, port, url=None, path=None): - super(PathSSH, self).__init__(url, path) - self.host = host - self.user = user - self.port = port - - def __str__(self): - if not self.url: - return urlunsplit( - ( - self.scheme, - "{}@{}:{}".format(self.user, self.host, self.port), - self.path, - "", - "", - ) - ) - return self.url diff --git a/dvc/path_info.py b/dvc/path_info.py new file mode 100644 index 000000000..a4c5a60d7 --- /dev/null +++ b/dvc/path_info.py @@ -0,0 +1,213 @@ +from __future__ import unicode_literals, absolute_import + +import sys +import os +import posixpath +from funcy import cached_property + +from dvc.utils.compat import str, builtin_str, basestring, is_py2 +from dvc.utils.compat import pathlib, fspath_py35, urlparse + + +# On Python 2.7/Windows sys.getfilesystemencoding() is set to mbcs, +# which is lossy, thus we can't use that, +# see https://github.com/mcmtroffaes/pathlib2/issues/56. +if is_py2: + fs_encoding = "utf-8" + + +class PathInfo(pathlib.PurePath): + scheme = "local" + + def __new__(cls, *args): + # Construct a proper subclass depending on current os + if cls is PathInfo: + cls = WindowsPathInfo if os.name == "nt" else PosixPathInfo + return cls._from_parts(args) + + @classmethod + def from_posix(cls, s): + return cls(PosixPathInfo(s)) + + def __str__(self): + path = self.__fspath__() + # Windows path on different drive than curdir doesn't have relpath + if os.name == "nt" and not os.path.commonprefix([os.getcwd(), path]): + return path + return os.path.relpath(path) + + def __repr__(self): + return builtin_str("{}: '{}'").format(type(self).__name__, self) + + # This permits passing it to file utils directly in Python 3.6+ + # With Python 2.7, Python 3.5+ we are stuck with path_info.fspath for now + def __fspath__(self): + return pathlib.PurePath.__str__(self) + + @property + def fspath(self): + return self.__fspath__() + + url = fspath + + def relpath(self, other): + return self.__class__( + os.path.relpath(fspath_py35(self), fspath_py35(other)) + ) + + def isin(self, other): + if isinstance(other, basestring): + other = self.__class__(other) + elif self.__class__ != other.__class__: + return False + return any(p == other for p in self.parents) + + # pathlib2 uses bytes internally in Python 2, and we use unicode everywhere + # for paths in both pythons, thus we need this glue. + if is_py2: + __unicode__ = __str__ + + def __str__(self): + return self.__unicode__().encode(sys.getfilesystemencoding()) + + @classmethod + def _parse_args(cls, args): + args = [ + a.encode(fs_encoding) + if isinstance(a, unicode) # noqa: F821 + else a + for a in args + ] + return super(PathInfo, cls)._parse_args(args) + + @property + def name(self): + return super(PathInfo, self).name.decode(fs_encoding) + + def __fspath__(self): # noqa: F811 + return pathlib.PurePath.__str__(self).decode(fs_encoding) + + def with_name(self, name): + return pathlib.PurePath.with_name(self, name.encode(fs_encoding)) + + def as_posix(self): + f = self._flavour + return str(self).replace(f.sep, "/") + + +class WindowsPathInfo(PathInfo, pathlib.PureWindowsPath): + pass + + +class PosixPathInfo(PathInfo, pathlib.PurePosixPath): + pass + + +class URLInfo(object): + DEFAULT_PORTS = {"http": 80, "https": 443, "ssh": 22} + + def __init__(self, url): + self.parsed = urlparse(url) + assert self.parsed.scheme != "remote" + + @classmethod + def from_parts( + cls, scheme=None, netloc=None, host=None, user=None, port=None, path="" + ): + assert scheme and (bool(host) ^ bool(netloc)) + + if netloc is None: + netloc = host + if user: + netloc = user + "@" + host + if port: + netloc += ":" + str(port) + return cls("{}://{}{}".format(scheme, netloc, path)) + + @cached_property + def url(self): + p = self.parsed + netloc = p.hostname + if p.username: + netloc = p.username + "@" + netloc + if p.port and int(p.port) != self.DEFAULT_PORTS.get(p.scheme): + netloc += ":" + str(p.port) + return "{}://{}{}".format(p.scheme, netloc, p.path) + + def __str__(self): + return self.url + + def __repr__(self): + return "{}: '{}'".format(type(self).__name__, self) + + def __eq__(self, other): + if isinstance(other, basestring): + other = self.__class__(other) + return str(self) == str(other) + + def __hash__(self): + return hash(self.url) + + def __div__(self, other): + p = self.parsed + new_path = posixpath.join(p.path, str(other)) + if not new_path.startswith("/"): + new_path = "/" + new_path + new_url = "{}://{}{}".format(p.scheme, p.netloc, new_path) + return self.__class__(new_url) + + __truediv__ = __div__ + + def __getattr__(self, name): + return getattr(self.parsed, name) + + @property + def port(self): + return self.parsed.port or self.DEFAULT_PORTS.get(self.parsed.scheme) + + @property + def host(self): + return self.parsed.hostname + + @property + def user(self): + return self.parsed.username + + @cached_property + def _path(self): + return pathlib.PurePosixPath(self.parsed.path) + + @property + def name(self): + return self._path.name + + @property + def parts(self): + return (self.scheme, self.netloc) + self._path.parts + + @property + def bucket(self): + return self.netloc + + @property + def parent(self): + return self.from_parts( + scheme=self.scheme, netloc=self.netloc, path=self._path.parent + ) + + def isin(self, other): + if isinstance(other, basestring): + other = self.__class__(other) + elif self.__class__ != other.__class__: + return False + return ( + self.scheme == other.scheme + and self.netloc == other.netloc + and PathInfo(self.path).isin(PathInfo(other.path)) + ) + + +class CloudURLInfo(URLInfo): + @property + def path(self): + return self.parsed.path.lstrip("/") diff --git a/dvc/remote/azure.py b/dvc/remote/azure.py index df86d7540..b31c65c5c 100644 --- a/dvc/remote/azure.py +++ b/dvc/remote/azure.py @@ -6,7 +6,6 @@ import re import logging from dvc.scheme import Schemes -from dvc.path.azure import PathAZURE try: from azure.storage.blob import BlockBlobService @@ -15,10 +14,11 @@ except ImportError: BlockBlobService = None from dvc.utils import tmp_fname, move -from dvc.utils.compat import urlparse, makedirs +from dvc.utils.compat import urlparse, makedirs, fspath_py35 from dvc.progress import progress from dvc.config import Config from dvc.remote.base import RemoteBASE +from dvc.path_info import CloudURLInfo logger = logging.getLogger(__name__) @@ -34,6 +34,7 @@ class Callback(object): class RemoteAZURE(RemoteBASE): scheme = Schemes.AZURE + path_cls = CloudURLInfo REGEX = ( r"azure://((?P[^=;]*)?|(" # backward compatibility @@ -43,62 +44,62 @@ class RemoteAZURE(RemoteBASE): REQUIRES = {"azure-storage-blob": BlockBlobService} PARAM_CHECKSUM = "etag" COPY_POLL_SECONDS = 5 + SUPPORTS_NO_TRAVERSE = False def __init__(self, repo, config): super(RemoteAZURE, self).__init__(repo, config) - self.url = config.get(Config.SECTION_REMOTE_URL, "azure://") - match = re.match(self.REGEX, self.url) # backward compatibility + url = config.get(Config.SECTION_REMOTE_URL, "azure://") + match = re.match(self.REGEX, url) # backward compatibility path = match.group("path") - self.bucket = ( - urlparse(self.url if path else "").netloc + bucket = ( + urlparse(url if path else "").netloc or match.group("container_name") # backward compatibility or os.getenv("AZURE_STORAGE_CONTAINER_NAME") ) - self.prefix = urlparse(self.url).path.lstrip("/") if path else "" - self.connection_string = ( config.get(Config.SECTION_AZURE_CONNECTION_STRING) or match.group("connection_string") # backward compatibility or os.getenv("AZURE_STORAGE_CONNECTION_STRING") ) - if not self.bucket: + if not bucket: raise ValueError("azure storage container name missing") if not self.connection_string: raise ValueError("azure storage connection string missing") self.__blob_service = None - self.path_info = PathAZURE(bucket=self.bucket) + self.path_info = ( + self.path_cls(url) + if path + else self.path_cls.from_parts(scheme=self.scheme, netloc=bucket) + ) @property def blob_service(self): if self.__blob_service is None: - logger.debug("URL {}".format(self.url)) + logger.debug("URL {}".format(self.path_info)) logger.debug("Connection string {}".format(self.connection_string)) self.__blob_service = BlockBlobService( connection_string=self.connection_string ) - logger.debug("Container name {}".format(self.bucket)) + logger.debug("Container name {}".format(self.path_info.bucket)) try: # verify that container exists self.__blob_service.list_blobs( - self.bucket, delimiter="/", num_results=1 + self.path_info.bucket, delimiter="/", num_results=1 ) except AzureMissingResourceHttpError: - self.__blob_service.create_container(self.bucket) + self.__blob_service.create_container(self.path_info.bucket) return self.__blob_service def remove(self, path_info): if path_info.scheme != self.scheme: raise NotImplementedError - logger.debug( - "Removing azure://{}/{}".format(path_info.bucket, path_info.path) - ) - + logger.debug("Removing {}".format(path_info)) self.blob_service.delete_blob(path_info.bucket, path_info.path) def _list_paths(self, bucket, prefix): @@ -118,7 +119,7 @@ class RemoteAZURE(RemoteBASE): next_marker = blobs.next_marker def list_cache_paths(self): - return self._list_paths(self.bucket, self.prefix) + return self._list_paths(self.path_info.bucket, self.path_info.path) def upload(self, from_infos, to_infos, names=None, no_progress_bar=False): names = self._verify_path_args(to_infos, from_infos, names) @@ -130,26 +131,21 @@ class RemoteAZURE(RemoteBASE): if from_info.scheme != "local": raise NotImplementedError - bucket = to_info.bucket - path = to_info.path - - logger.debug( - "Uploading '{}' to '{}/{}'".format( - from_info.path, bucket, path - ) - ) - + logger.debug("Uploading '{}' to '{}'".format(from_info, to_info)) if not name: - name = os.path.basename(from_info.path) + name = from_info.name cb = None if no_progress_bar else Callback(name) try: self.blob_service.create_blob_from_path( - bucket, path, from_info.path, progress_callback=cb + to_info.bucket, + to_info.path, + from_info.fspath, + progress_callback=cb, ) except Exception: - msg = "failed to upload '{}'".format(from_info.path) + msg = "failed to upload '{}'".format(from_info) logger.warning(msg) else: progress.finish_target(name) @@ -171,32 +167,28 @@ class RemoteAZURE(RemoteBASE): if to_info.scheme != "local": raise NotImplementedError - bucket = from_info.bucket - path = from_info.path + logger.debug("Downloading '{}' to '{}'".format(from_info, to_info)) - logger.debug( - "Downloading '{}/{}' to '{}'".format( - bucket, path, to_info.path - ) - ) - - tmp_file = tmp_fname(to_info.path) + tmp_file = tmp_fname(to_info) if not name: - name = os.path.basename(to_info.path) + name = to_info.name cb = None if no_progress_bar else Callback(name) - makedirs(os.path.dirname(to_info.path), exist_ok=True) + makedirs(fspath_py35(to_info.parent), exist_ok=True) try: self.blob_service.get_blob_to_path( - bucket, path, tmp_file, progress_callback=cb + from_info.bucket, + from_info.path, + tmp_file, + progress_callback=cb, ) except Exception: - msg = "failed to download '{}/{}'".format(bucket, path) + msg = "failed to download '{}'".format(from_info) logger.warning(msg) else: - move(tmp_file, to_info.path) + move(tmp_file, to_info.fspath) if not no_progress_bar: progress.finish_target(name) diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 30f13a8f4..8f1be8257 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -1,26 +1,23 @@ from __future__ import unicode_literals -from copy import copy - -from dvc.path.local import PathLOCAL -from dvc.utils.compat import str +from dvc.utils.compat import str, basestring, urlparse import os -import re import json -import ntpath import logging import tempfile -import posixpath +import itertools from operator import itemgetter from multiprocessing import cpu_count +from concurrent.futures import ThreadPoolExecutor import dvc.prompt as prompt from dvc.config import Config from dvc.exceptions import DvcException, ConfirmRemoveError from dvc.progress import progress -from dvc.utils import LARGE_DIR_SIZE, tmp_fname +from dvc.utils import LARGE_DIR_SIZE, tmp_fname, to_chunks from dvc.state import StateBase +from dvc.path_info import PathInfo, URLInfo logger = logging.getLogger(__name__) @@ -67,9 +64,10 @@ class RemoteMissingDepsError(DvcException): class RemoteBASE(object): - scheme = None - REGEX = None + scheme = "base" + path_cls = URLInfo REQUIRES = {} + SUPPORTS_NO_TRAVERSE = True JOBS = 4 * cpu_count() PARAM_RELPATH = "relpath" @@ -104,13 +102,14 @@ class RemoteBASE(object): raise RemoteMissingDepsError(msg) self.protected = False + self.no_traverse = config.get(Config.SECTION_REMOTE_NO_TRAVERSE) self.state = StateBase() - self._dir_info = {} def __repr__(self): - return "{class_name}: '{url}'".format( - class_name=type(self).__name__, url=(self.url or "No url") + return "{class_name}: '{path_info}'".format( + class_name=type(self).__name__, + path_info=self.path_info or "No path", ) def compat_config(config): @@ -118,39 +117,14 @@ class RemoteBASE(object): @classmethod def supported(cls, config): - url = config[Config.SECTION_REMOTE_URL] - return cls.match(url) is not None + if isinstance(config, basestring): + url = config + else: + url = config[Config.SECTION_REMOTE_URL] - @staticmethod - def _replace_sep(path, from_sep, to_sep): - assert not ntpath.isabs(path) - assert not posixpath.isabs(path) - return path.replace(from_sep, to_sep) - - @classmethod - def to_posixpath(cls, path): - return cls._replace_sep(path, "\\", "/") - - @classmethod - def to_ntpath(cls, path): - return cls._replace_sep(path, "/", "\\") - - def to_ospath(self, path): - if self.ospath == ntpath: - return self.to_ntpath(path) - elif self.ospath == posixpath: - return self.to_posixpath(path) - assert False - - @classmethod - def match(cls, url): - return re.match(cls.REGEX, url) - - def group(self, name): - m = self.match(self.url) - if not m: - return None - return m.group(name) + # NOTE: silently skipping remote, calling code should handle that + parsed = urlparse(url) + return parsed.scheme == cls.scheme @property def cache(self): @@ -162,28 +136,31 @@ class RemoteBASE(object): def _collect_dir(self, path_info): dir_info = [] - p_info = copy(path_info) - dpath = p_info.path for root, dirs, files in self.walk(path_info): if len(files) > LARGE_DIR_SIZE: msg = ( "Computing md5 for a large directory {}. " "This is only done once." ) - relpath = self.ospath.relpath(root) - logger.info(msg.format(relpath)) - files = progress(files, name=relpath) + title = str(self.path_cls(root)) + logger.info(msg.format(title)) + files = progress(files, name=title) for fname in files: - path = self.ospath.join(root, fname) - p_info.path = path - relpath = self.to_posixpath(self.ospath.relpath(path, dpath)) - - checksum = self.get_file_checksum(p_info) + file_info = self.path_cls(root) / fname + relpath = file_info.relative_to(path_info) dir_info.append( { - self.PARAM_RELPATH: relpath, - self.PARAM_CHECKSUM: checksum, + # NOTE: this is lossy transformation: + # "hey\there" -> "hey/there" + # "hey/there" -> "hey/there" + # The latter is fine filename on Windows, + # which will transform to dir/file on back transform. + # + # Yes, this is a BUG, as long as we permit "/" in + # filenames on Windows and "\" on Unix + self.PARAM_RELPATH: relpath.as_posix(), + self.PARAM_CHECKSUM: self.get_file_checksum(file_info), } ) @@ -192,32 +169,27 @@ class RemoteBASE(object): def get_dir_checksum(self, path_info): dir_info = self._collect_dir(path_info) - checksum, from_info, to_info = self._get_dir_info_checksum( - dir_info, path_info - ) + checksum, tmp_info = self._get_dir_info_checksum(dir_info) + new_info = self.cache.checksum_to_path_info(checksum) if self.cache.changed_cache_file(checksum): - self.cache.move(from_info, to_info) + self.cache.move(tmp_info, new_info) self.state.save(path_info, checksum) - self.state.save(to_info, checksum) + self.state.save(new_info, checksum) return checksum - def _get_dir_info_checksum(self, dir_info, path_info): - to_info = copy(path_info) - to_info.path = self.cache.ospath.join(self.cache.prefix, tmp_fname("")) - + def _get_dir_info_checksum(self, dir_info): tmp = tempfile.NamedTemporaryFile(delete=False).name with open(tmp, "w+") as fobj: json.dump(dir_info, fobj, sort_keys=True) - from_info = PathLOCAL(path=tmp) + from_info = PathInfo(tmp) + to_info = self.cache.path_info / tmp_fname("") self.cache.upload([from_info], [to_info], no_progress_bar=True) checksum = self.get_file_checksum(to_info) + self.CHECKSUM_DIR_SUFFIX - from_info = copy(to_info) - to_info.path = self.cache.checksum_to_path(checksum) - return checksum, from_info, to_info + return checksum, to_info def get_dir_cache(self, checksum): assert checksum @@ -235,7 +207,7 @@ class RemoteBASE(object): fobj = tempfile.NamedTemporaryFile(delete=False) path = fobj.name - to_info = PathLOCAL(path=path) + to_info = self.path_cls(path) self.cache.download([path_info], [to_info], no_progress_bar=True) try: @@ -253,7 +225,9 @@ class RemoteBASE(object): return [] for info in d: - info[self.PARAM_RELPATH] = self.to_ospath(info[self.PARAM_RELPATH]) + # NOTE: here is a BUG, see comment to .as_posix() below + relpath = self.path_cls.from_posix(info[self.PARAM_RELPATH]) + info[self.PARAM_RELPATH] = relpath.fspath return d @@ -333,7 +307,7 @@ class RemoteBASE(object): logger.debug("'{}' hasn't changed.".format(path_info)) return False - def link(self, from_info, to_info): + def link(self, from_info, to_info, link_type=None): self.copy(from_info, to_info) def _save_file(self, path_info, checksum, save_link=True): @@ -360,13 +334,9 @@ class RemoteBASE(object): cache_info = self.checksum_to_path_info(checksum) dir_info = self.get_dir_cache(checksum) - entry_info = copy(path_info) for entry in dir_info: + entry_info = path_info / entry[self.PARAM_RELPATH] entry_checksum = entry[self.PARAM_CHECKSUM] - entry_info.path = self.ospath.join( - path_info.path, entry[self.PARAM_RELPATH] - ) - self._save_file(entry_info, entry_checksum, save_link=False) self.state.save_link(path_info) @@ -385,7 +355,8 @@ class RemoteBASE(object): def walk(self, path_info): raise NotImplementedError - def protect(self, path_info): + @staticmethod + def protect(path_info): pass def save(self, path_info, checksum_info): @@ -450,25 +421,11 @@ class RemoteBASE(object): return names - @property - def ospath(self): - return posixpath - - def checksum_to_path(self, checksum): - assert checksum - return self.ospath.join(self.prefix, checksum[0:2], checksum[2:]) - def path_to_checksum(self, path): - relpath = self.ospath.relpath(path, self.prefix) - return self.ospath.dirname(relpath) + self.ospath.basename(relpath) + return "".join(self.path_cls(path).parts[-2:]) def checksum_to_path_info(self, checksum): - path_info = copy(self.path_info) - path_info.path = self.checksum_to_path(checksum) - return path_info - - def md5s_to_path_infos(self, md5s): - return [self.checksum_to_path_info(md5) for md5 in md5s] + return self.path_info / checksum[0:2] / checksum[2:] def list_cache_paths(self): raise NotImplementedError @@ -483,9 +440,7 @@ class RemoteBASE(object): ) def gc(self, cinfos): - from dvc.remote.local import RemoteLOCAL - - used = {info[RemoteLOCAL.PARAM_CHECKSUM] for info in cinfos["local"]} + used = self.extract_used_local_checksums(cinfos) if self.scheme != "": used |= { @@ -539,11 +494,15 @@ class RemoteBASE(object): if self.changed_cache_file(checksum): return True + if not self._changed_unpacked_dir(checksum): + return False + for entry in self.get_dir_cache(checksum): - checksum = entry[self.PARAM_CHECKSUM] - if self.changed_cache_file(checksum): + entry_checksum = entry[self.PARAM_CHECKSUM] + if self.changed_cache_file(entry_checksum): return True + self._update_unpacked_dir(checksum) return False def changed_cache(self, checksum): @@ -552,15 +511,36 @@ class RemoteBASE(object): return self.changed_cache_file(checksum) def cache_exists(self, checksums): - # NOTE: The reason for such an odd logic is that most of the remotes - # take much shorter time to just retrieve everything they have under - # a certain prefix(e.g. s3, gs, ssh, hdfs). Other remotes that can - # check if particular file exists much quicker, use their own - # implementation of cache_exists(see http, local). - # - # Result of all() might be way too big, so we should walk through - # it in one pass. - return list(filter(lambda checksum: checksum in checksums, self.all())) + """Check if the given checksums are stored in the remote. + + There are two ways of performing this check: + + - Traverse: Get a list of all the files in the remote + (traversing the cache directory) and compare it with + the given checksums. + + - No traverse: For each given checksum, run the `exists` + method and filter the checksums that aren't on the remote. + This is done in parallel threads. + + The reason for such an odd logic is that most of the remotes + take much shorter time to just retrieve everything they have under + a certain prefix (e.g. s3, gs, ssh, hdfs). Other remotes that can + check if particular file exists much quicker, use their own + implementation of cache_exists (see http, local). + + Returns: + A list with checksums that were found in the remote + """ + if self.no_traverse and self.SUPPORTS_NO_TRAVERSE: + with ThreadPoolExecutor(max_workers=self.JOBS) as executor: + path_infos = [self.checksum_to_path_info(x) for x in checksums] + chunks = to_chunks(path_infos, self.JOBS) + results = executor.map(self.exists, chunks) + in_remote = itertools.chain.from_iterable(results) + return list(itertools.compress(checksums, in_remote)) + + return list(set(checksums) & set(self.all())) def already_cached(self, path_info): current = self.get_checksum(path_info) @@ -598,7 +578,7 @@ class RemoteBASE(object): self.state.save_link(path_info) self.state.save(path_info, checksum) if progress_callback: - progress_callback.update(path_info.url) + progress_callback.update(str(path_info)) def makedirs(self, path_info): raise NotImplementedError @@ -615,13 +595,11 @@ class RemoteBASE(object): logger.debug("Linking directory '{}'.".format(path_info)) - entry_info = copy(path_info) for entry in dir_info: relpath = entry[self.PARAM_RELPATH] entry_checksum = entry[self.PARAM_CHECKSUM] entry_cache_info = self.checksum_to_path_info(entry_checksum) - entry_info.url = self.ospath.join(path_info.url, relpath) - entry_info.path = self.ospath.join(path_info.path, relpath) + entry_info = path_info / relpath entry_checksum_info = {self.PARAM_CHECKSUM: entry_checksum} if self.changed(entry_info, entry_checksum_info): @@ -630,7 +608,7 @@ class RemoteBASE(object): self.link(entry_cache_info, entry_info) self.state.save(entry_info, entry_checksum) if progress_callback: - progress_callback.update(entry_info.url) + progress_callback.update(str(entry_info)) self._remove_redundant_files(path_info, dir_info, force) @@ -639,30 +617,22 @@ class RemoteBASE(object): def _remove_redundant_files(self, path_info, dir_info, force): existing_files = set( - self.ospath.join(root, fname) + self.path_cls(root) / fname for root, _, files in self.walk(path_info) for fname in files ) - needed_files = set( - self.ospath.join(path_info.path, entry[self.PARAM_RELPATH]) - for entry in dir_info - ) + needed_files = { + path_info / entry[self.PARAM_RELPATH] for entry in dir_info + } - delta = existing_files - needed_files - - d_info = copy(path_info) - for path in delta: - d_info.path = path - self.safe_remove(d_info, force) + for path in existing_files - needed_files: + self.safe_remove(path, force) def checkout( self, path_info, checksum_info, force=False, progress_callback=None ): - if ( - path_info.scheme not in ["local"] - and path_info.scheme != self.scheme - ): + if path_info.scheme not in ["local", self.scheme]: raise NotImplementedError checksum = checksum_info.get(self.PARAM_CHECKSUM) @@ -701,3 +671,19 @@ class RemoteBASE(object): @staticmethod def unprotect(path_info): pass + + def _get_unpacked_dir_names(self, checksums): + return set() + + def extract_used_local_checksums(self, cinfos): + from dvc.remote import RemoteLOCAL + + used = {info[RemoteLOCAL.PARAM_CHECKSUM] for info in cinfos["local"]} + unpacked = self._get_unpacked_dir_names(used) + return used | unpacked + + def _changed_unpacked_dir(self, checksum): + return True + + def _update_unpacked_dir(self, checksum, progress_callback=None): + pass diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index 2e747f1dd..519da7962 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -1,10 +1,9 @@ from __future__ import unicode_literals -import os import logging +import itertools from dvc.scheme import Schemes -from dvc.path.gs import PathGS try: from google.cloud import storage @@ -12,11 +11,12 @@ except ImportError: storage = None from dvc.utils import tmp_fname, move -from dvc.utils.compat import urlparse, makedirs +from dvc.utils.compat import makedirs, fspath_py35 from dvc.remote.base import RemoteBASE from dvc.config import Config from dvc.progress import progress from dvc.exceptions import DvcException +from dvc.path_info import CloudURLInfo logger = logging.getLogger(__name__) @@ -24,29 +24,24 @@ logger = logging.getLogger(__name__) class RemoteGS(RemoteBASE): scheme = Schemes.GS - REGEX = r"^gs://(?P.*)$" + path_cls = CloudURLInfo REQUIRES = {"google.cloud.storage": storage} PARAM_CHECKSUM = "md5" def __init__(self, repo, config): super(RemoteGS, self).__init__(repo, config) - storagepath = "gs://" - storagepath += config.get(Config.SECTION_AWS_STORAGEPATH, "/") - storagepath.lstrip("/") - self.url = config.get(Config.SECTION_REMOTE_URL, storagepath) + + storagepath = "gs://" + config.get(Config.SECTION_GCP_STORAGEPATH, "/") + url = config.get(Config.SECTION_REMOTE_URL, storagepath) + self.path_info = self.path_cls(url) + self.projectname = config.get(Config.SECTION_GCP_PROJECTNAME, None) self.credentialpath = config.get(Config.SECTION_GCP_CREDENTIALPATH) - parsed = urlparse(self.url) - self.bucket = parsed.netloc - self.prefix = parsed.path.lstrip("/") - - self.path_info = PathGS(bucket=self.bucket) - @staticmethod def compat_config(config): ret = config.copy() - url = "gs://" + ret.pop(Config.SECTION_AWS_STORAGEPATH, "").lstrip("/") + url = "gs://" + ret.pop(Config.SECTION_GCP_STORAGEPATH, "").lstrip("/") ret[Config.SECTION_REMOTE_URL] = url return ret @@ -89,29 +84,44 @@ class RemoteGS(RemoteBASE): if path_info.scheme != "gs": raise NotImplementedError - logger.debug( - "Removing gs://{}/{}".format(path_info.bucket, path_info.path) - ) - + logger.debug("Removing gs://{}".format(path_info)) blob = self.gs.bucket(path_info.bucket).get_blob(path_info.path) if not blob: return blob.delete() - def _list_paths(self, bucket, prefix): - for blob in self.gs.bucket(bucket).list_blobs(prefix=prefix): + def _list_paths(self, bucket, prefix, gs=None): + gs = gs or self.gs + + for blob in gs.bucket(bucket).list_blobs(prefix=prefix): yield blob.name def list_cache_paths(self): - return self._list_paths(self.bucket, self.prefix) + return self._list_paths(self.path_info.bucket, self.path_info.path) - def exists(self, path_info): - assert not isinstance(path_info, list) - assert path_info.scheme == "gs" + def exists(self, path_infos): + single_path = False - paths = self._list_paths(path_info.bucket, path_info.path) - return any(path_info.path == path for path in paths) + if not isinstance(path_infos, list): + single_path = True + path_infos = [path_infos] + + gs = self.gs + + paths = itertools.chain.from_iterable( + self._list_paths(path_info.bucket, path_info.path, gs) + for path_info in path_infos + ) + + paths = set(paths) + + results = [path_info.path in paths for path_info in path_infos] + + if single_path and results: + return all(results) + + return results def upload(self, from_infos, to_infos, names=None, no_progress_bar=False): names = self._verify_path_args(to_infos, from_infos, names) @@ -125,14 +135,10 @@ class RemoteGS(RemoteBASE): if from_info.scheme != "local": raise NotImplementedError - logger.debug( - "Uploading '{}' to '{}/{}'".format( - from_info.path, to_info.bucket, to_info.path - ) - ) + logger.debug("Uploading '{}' to '{}'".format(from_info, to_info)) if not name: - name = os.path.basename(from_info.path) + name = from_info.name if not no_progress_bar: progress.update_target(name, 0, None) @@ -140,12 +146,10 @@ class RemoteGS(RemoteBASE): try: bucket = gs.bucket(to_info.bucket) blob = bucket.blob(to_info.path) - blob.upload_from_filename(from_info.path) + blob.upload_from_filename(from_info.fspath) except Exception: - msg = "failed to upload '{}' to '{}/{}'" - logger.exception( - msg.format(from_info.path, to_info.bucket, to_info.path) - ) + msg = "failed to upload '{}' to '{}'" + logger.exception(msg.format(from_info, to_info)) continue progress.finish_target(name) @@ -173,34 +177,29 @@ class RemoteGS(RemoteBASE): if to_info.scheme != "local": raise NotImplementedError - msg = "Downloading '{}/{}' to '{}'".format( - from_info.bucket, from_info.path, to_info.path - ) + msg = "Downloading '{}' to '{}'".format(from_info, to_info) logger.debug(msg) - tmp_file = tmp_fname(to_info.path) + tmp_file = tmp_fname(to_info) if not name: - name = os.path.basename(to_info.path) + name = to_info.name if not no_progress_bar: # percent_cb is not available for download_to_filename, so # lets at least update progress at pathpoints(start, finish) progress.update_target(name, 0, None) - makedirs(os.path.dirname(to_info.path), exist_ok=True) - + makedirs(fspath_py35(to_info.parent), exist_ok=True) try: bucket = gs.bucket(from_info.bucket) blob = bucket.get_blob(from_info.path) blob.download_to_filename(tmp_file) except Exception: - msg = "failed to download '{}/{}' to '{}'" - logger.exception( - msg.format(from_info.bucket, from_info.path, to_info.path) - ) + msg = "failed to download '{}' to '{}'" + logger.exception(msg.format(from_info, to_info)) continue - move(tmp_file, to_info.path) + move(tmp_file, fspath_py35(to_info)) if not no_progress_bar: progress.finish_target(name) diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py index b8a3b22cc..206d61bda 100644 --- a/dvc/remote/hdfs.py +++ b/dvc/remote/hdfs.py @@ -9,7 +9,7 @@ from subprocess import Popen, PIPE from dvc.config import Config from dvc.scheme import Schemes -from dvc.path.hdfs import PathHDFS + from dvc.remote.base import RemoteBASE, RemoteCmdError from dvc.utils import fix_env, tmp_fname @@ -20,20 +20,20 @@ logger = logging.getLogger(__name__) class RemoteHDFS(RemoteBASE): scheme = Schemes.HDFS REGEX = r"^hdfs://((?P.*)@)?.*$" + SUPPORTS_NO_TRAVERSE = False PARAM_CHECKSUM = "checksum" def __init__(self, repo, config): super(RemoteHDFS, self).__init__(repo, config) - self.url = config.get(Config.SECTION_REMOTE_URL, "/") - self.prefix = self.url - self.user = self.group("user") + url = config.get(Config.SECTION_REMOTE_URL, "/") + self.path_info = self.path_cls(url) + + self.user = self.path_info.user if not self.user: self.user = config.get( Config.SECTION_REMOTE_USER, getpass.getuser() ) - self.path_info = PathHDFS(user=self.user) - def hadoop_fs(self, cmd, user=None): cmd = "hadoop fs -" + cmd if user: @@ -114,19 +114,18 @@ class RemoteHDFS(RemoteBASE): raise NotImplementedError self.hadoop_fs( - "mkdir -p {}".format(posixpath.dirname(to_info.path)), - user=to_info.user, + "mkdir -p {}".format(to_info.parent.url), user=to_info.user ) - tmp_file = tmp_fname(to_info.path) + tmp_file = tmp_fname(to_info.url) self.hadoop_fs( - "copyFromLocal {} {}".format(from_info.path, tmp_file), + "copyFromLocal {} {}".format(from_info.fspath, tmp_file), user=to_info.user, ) self.hadoop_fs( - "mv {} {}".format(tmp_file, to_info.path), user=to_info.user + "mv {} {}".format(tmp_file, to_info.url), user=to_info.user ) def download( @@ -150,26 +149,26 @@ class RemoteHDFS(RemoteBASE): if to_info.scheme != "local": raise NotImplementedError - dname = os.path.dirname(to_info.path) + dname = to_info.parent.fspath if not os.path.exists(dname): os.makedirs(dname) - tmp_file = tmp_fname(to_info.path) + tmp_file = tmp_fname(to_info.fspath) self.hadoop_fs( - "copyToLocal {} {}".format(from_info.path, tmp_file), + "copyToLocal {} {}".format(from_info.url, tmp_file), user=from_info.user, ) - os.rename(tmp_file, to_info.path) + os.rename(tmp_file, to_info.fspath) def list_cache_paths(self): try: - self.hadoop_fs("test -e {}".format(self.prefix)) + self.hadoop_fs("test -e {}".format(self.path_info.url)) except RemoteCmdError: return [] - stdout = self.hadoop_fs("ls -R {}".format(self.prefix)) + stdout = self.hadoop_fs("ls -R {}".format(self.path_info.url)) lines = stdout.split("\n") flist = [] for line in lines: diff --git a/dvc/remote/http.py b/dvc/remote/http.py index 6af541104..20498cbf3 100644 --- a/dvc/remote/http.py +++ b/dvc/remote/http.py @@ -1,8 +1,8 @@ from __future__ import unicode_literals from dvc.scheme import Schemes -from dvc.path import Path -from dvc.utils.compat import open, makedirs + +from dvc.utils.compat import open, makedirs, fspath_py35 import os import threading @@ -34,21 +34,15 @@ class ProgressBarCallback(object): class RemoteHTTP(RemoteBASE): scheme = Schemes.HTTP - REGEX = r"^http://.*$" REQUEST_TIMEOUT = 10 CHUNK_SIZE = 2 ** 16 PARAM_CHECKSUM = "etag" def __init__(self, repo, config): super(RemoteHTTP, self).__init__(repo, config) - self.cache_dir = config.get(Config.SECTION_REMOTE_URL) - self.url = self.cache_dir - self.path_info = Path(self.scheme) - - @property - def prefix(self): - return self.cache_dir + url = config.get(Config.SECTION_REMOTE_URL) + self.path_info = self.path_cls(url) if url else None def download( self, @@ -67,17 +61,15 @@ class RemoteHTTP(RemoteBASE): if to_info.scheme != "local": raise NotImplementedError - msg = "Downloading '{}' to '{}'".format( - from_info.path, to_info.path - ) + msg = "Downloading '{}' to '{}'".format(from_info, to_info) logger.debug(msg) if not name: - name = os.path.basename(to_info.path) + name = to_info.name - makedirs(os.path.dirname(to_info.path), exist_ok=True) + makedirs(fspath_py35(to_info.parent), exist_ok=True) - total = self._content_length(from_info.path) + total = self._content_length(from_info.url) if no_progress_bar or not total: cb = None @@ -86,27 +78,40 @@ class RemoteHTTP(RemoteBASE): try: self._download_to( - from_info.path, to_info.path, callback=cb, resume=resume + from_info.url, to_info.fspath, callback=cb, resume=resume ) except Exception: - msg = "failed to download '{}'".format(from_info.path) + msg = "failed to download '{}'".format(from_info) logger.exception(msg) continue if not no_progress_bar: progress.finish_target(name) - def exists(self, path_info): - assert not isinstance(path_info, list) - assert path_info.scheme == self.scheme - return bool(self._request("HEAD", path_info.path)) + def exists(self, path_infos): + single_path = False + + if not isinstance(path_infos, list): + single_path = True + path_infos = [path_infos] + + results = [ + bool(self._request("HEAD", path_info.url)) + for path_info in path_infos + ] + + if single_path and results: + return all(results) + + return results def cache_exists(self, md5s): assert isinstance(md5s, list) def func(md5): - return bool(self._request("HEAD", self.checksum_to_path(md5))) + url = self.checksum_to_path_info(md5).url + return bool(self._request("HEAD", url)) return list(filter(func, md5s)) @@ -114,7 +119,7 @@ class RemoteHTTP(RemoteBASE): return self._request("HEAD", url).headers.get("Content-Length") def get_file_checksum(self, path_info): - url = path_info.path + url = path_info.url etag = self._request("HEAD", url).headers.get("ETag") or self._request( "HEAD", url ).headers.get("Content-MD5") diff --git a/dvc/remote/local/__init__.py b/dvc/remote/local/__init__.py index 3e314da4f..e33a59851 100644 --- a/dvc/remote/local/__init__.py +++ b/dvc/remote/local/__init__.py @@ -3,17 +3,13 @@ from __future__ import unicode_literals from copy import copy from dvc.scheme import Schemes -from dvc.path.base import PathBASE -from dvc.path.local import PathLOCAL from dvc.remote.local.slow_link_detection import slow_link_guard -from dvc.utils.compat import str, makedirs +from dvc.utils.compat import str, makedirs, fspath_py35 import os import stat import uuid -import ntpath import shutil -import posixpath import logging from dvc.system import System @@ -38,16 +34,19 @@ from dvc.exceptions import DvcException from dvc.progress import progress from concurrent.futures import ThreadPoolExecutor +from dvc.path_info import PathInfo logger = logging.getLogger(__name__) class RemoteLOCAL(RemoteBASE): scheme = Schemes.LOCAL - REGEX = r"^(?P.*)$" + path_cls = PathInfo PARAM_CHECKSUM = "md5" PARAM_PATH = "path" + UNPACKED_DIR_SUFFIX = ".unpacked" + DEFAULT_CACHE_TYPES = ["reflink", "copy"] CACHE_TYPE_MAP = { "copy": shutil.copyfile, @@ -60,12 +59,6 @@ class RemoteLOCAL(RemoteBASE): super(RemoteLOCAL, self).__init__(repo, config) self.state = self.repo.state if self.repo else None self.protected = config.get(Config.SECTION_CACHE_PROTECTED, False) - storagepath = config.get(Config.SECTION_AWS_STORAGEPATH, None) - self.cache_dir = config.get(Config.SECTION_REMOTE_URL, storagepath) - - if self.cache_dir is not None and not os.path.isabs(self.cache_dir): - cwd = config[Config.PRIVATE_CWD] - self.cache_dir = os.path.abspath(os.path.join(cwd, self.cache_dir)) types = config.get(Config.SECTION_CACHE_TYPE, None) if types: @@ -75,36 +68,46 @@ class RemoteLOCAL(RemoteBASE): else: self.cache_types = copy(self.DEFAULT_CACHE_TYPES) - if self.cache_dir is not None and not os.path.exists(self.cache_dir): - os.mkdir(self.cache_dir) + # A clunky way to detect cache dir + storagepath = config.get(Config.SECTION_LOCAL_STORAGEPATH, None) + cache_dir = config.get(Config.SECTION_REMOTE_URL, storagepath) + if cache_dir is not None and not os.path.isabs(cache_dir): + cwd = config[Config.PRIVATE_CWD] + cache_dir = os.path.abspath(os.path.join(cwd, cache_dir)) + + if cache_dir is not None and not os.path.exists(cache_dir): + os.mkdir(cache_dir) + + self.path_info = PathInfo(cache_dir) if cache_dir else None self._dir_info = {} - self.path_info = PathLOCAL() + + @property + def cache_dir(self): + return self.path_info.fspath if self.path_info else None + + @classmethod + def supported(cls, config): + return True @staticmethod def compat_config(config): ret = config.copy() - url = ret.pop(Config.SECTION_AWS_STORAGEPATH, "") + url = ret.pop(Config.SECTION_LOCAL_STORAGEPATH, "") ret[Config.SECTION_REMOTE_URL] = url return ret - @property - def url(self): - return self.cache_dir - - @property - def prefix(self): - return self.cache_dir - def list_cache_paths(self): - clist = [] - for entry in os.listdir(self.cache_dir): - subdir = os.path.join(self.cache_dir, entry) - if not os.path.isdir(subdir): - continue + assert self.path_info is not None - for cache in os.listdir(subdir): - clist.append(os.path.join(subdir, cache)) + clist = [] + for entry in os.listdir(fspath_py35(self.path_info)): + subdir = self.path_info / entry + if not os.path.isdir(fspath_py35(subdir)): + continue + clist.extend( + subdir / cache for cache in os.listdir(fspath_py35(subdir)) + ) return clist @@ -112,68 +115,84 @@ class RemoteLOCAL(RemoteBASE): if not md5: return None - return self.checksum_to_path(md5) + return self.checksum_to_path_info(md5).url def exists(self, path_info): - assert isinstance(path_info, PathBASE) assert path_info.scheme == "local" - return os.path.lexists(path_info.path) + return os.path.lexists(fspath_py35(path_info)) def makedirs(self, path_info): if not self.exists(path_info): - os.makedirs(path_info.path) + os.makedirs(fspath_py35(path_info)) - @slow_link_guard - def link(self, cache_info, path_info): - cache = cache_info.path - path = path_info.path + def link(self, from_info, to_info, link_type=None): + from_path = from_info.fspath + to_path = to_info.fspath - assert os.path.isfile(cache) + assert os.path.isfile(from_path) - dname = os.path.dirname(path) + dname = os.path.dirname(to_path) if not os.path.exists(dname): os.makedirs(dname) # NOTE: just create an empty file for an empty cache - if os.path.getsize(cache) == 0: - open(path, "w+").close() + if os.path.getsize(from_path) == 0: + open(to_path, "w+").close() - msg = "Created empty file: {} -> {}".format(cache, path) + msg = "Created empty file: {} -> {}".format(from_path, to_path) logger.debug(msg) return - i = len(self.cache_types) + if not link_type: + link_types = self.cache_types + else: + link_types = [link_type] + + self._try_links(from_info, to_info, link_types) + + @classmethod + def _get_link_method(cls, link_type): + try: + return cls.CACHE_TYPE_MAP[link_type] + except KeyError: + raise DvcException( + "Cache type: '{}' not supported!".format(link_type) + ) + + def _link(self, from_info, to_info, link_method): + if self.exists(to_info): + raise DvcException("Link '{}' already exists!".format(to_info)) + else: + link_method(from_info.fspath, to_info.fspath) + + if self.protected: + self.protect(to_info) + + msg = "Created {}'{}': {} -> {}".format( + "protected " if self.protected else "", + self.cache_types[0], + from_info, + to_info, + ) + logger.debug(msg) + + @slow_link_guard + def _try_links(self, from_info, to_info, link_types): + i = len(link_types) while i > 0: + link_method = self._get_link_method(link_types[0]) try: - self.CACHE_TYPE_MAP[self.cache_types[0]](cache, path) - - if self.protected: - self.protect(path_info) - - msg = "Created {}'{}': {} -> {}".format( - "protected " if self.protected else "", - self.cache_types[0], - cache, - path, - ) - - logger.debug(msg) + self._link(from_info, to_info, link_method) return except DvcException as exc: msg = "Cache type '{}' is not supported: {}" - logger.debug(msg.format(self.cache_types[0], str(exc))) - del self.cache_types[0] + logger.debug(msg.format(link_types[0], str(exc))) + del link_types[0] i -= 1 raise DvcException("no possible cache types left to try out.") - @property - def ospath(self): - if os.name == "nt": - return ntpath - return posixpath - def already_cached(self, path_info): assert path_info.scheme in ["", "local"] @@ -185,7 +204,7 @@ class RemoteLOCAL(RemoteBASE): return not self.changed_cache(current_md5) def is_empty(self, path_info): - path = path_info.path + path = path_info.fspath if self.isfile(path_info) and os.path.getsize(path) == 0: return True @@ -196,29 +215,30 @@ class RemoteLOCAL(RemoteBASE): return False def isfile(self, path_info): - return os.path.isfile(path_info.path) + return os.path.isfile(fspath_py35(path_info)) def isdir(self, path_info): - return os.path.isdir(path_info.path) + return os.path.isdir(fspath_py35(path_info)) def walk(self, path_info): - return os.walk(path_info.path) + return os.walk(fspath_py35(path_info)) def get_file_checksum(self, path_info): - return file_md5(path_info.path)[0] + return file_md5(fspath_py35(path_info))[0] def remove(self, path_info): if path_info.scheme != "local": raise NotImplementedError - remove(path_info.path) + if self.exists(path_info): + remove(path_info.fspath) def move(self, from_info, to_info): if from_info.scheme != "local" or to_info.scheme != "local": raise NotImplementedError - inp = from_info.path - outp = to_info.path + inp = from_info.fspath + outp = to_info.fspath # moving in two stages to make the whole operation atomic in # case inp and outp are in different filesystems and actual @@ -228,8 +248,11 @@ class RemoteLOCAL(RemoteBASE): move(tmp, outp) def cache_exists(self, md5s): - assert isinstance(md5s, list) - return list(filter(lambda md5: not self.changed_cache_file(md5), md5s)) + return [ + checksum + for checksum in md5s + if not self.changed_cache_file(checksum) + ] def upload(self, from_infos, to_infos, names=None, no_progress_bar=False): names = self._verify_path_args(to_infos, from_infos, names) @@ -241,29 +264,25 @@ class RemoteLOCAL(RemoteBASE): if from_info.scheme != "local": raise NotImplementedError - logger.debug( - "Uploading '{}' to '{}'".format(from_info.path, to_info.path) - ) + logger.debug("Uploading '{}' to '{}'".format(from_info, to_info)) if not name: - name = os.path.basename(from_info.path) + name = from_info.name - makedirs(os.path.dirname(to_info.path), exist_ok=True) - tmp_file = tmp_fname(to_info.path) + makedirs(fspath_py35(to_info.parent), exist_ok=True) + tmp_file = tmp_fname(to_info) try: copyfile( - from_info.path, + fspath_py35(from_info), tmp_file, name=name, no_progress_bar=no_progress_bar, ) - os.rename(tmp_file, to_info.path) + os.rename(tmp_file, fspath_py35(to_info)) except Exception: logger.exception( - "failed to upload '{}' to '{}'".format( - from_info.path, to_info.path - ) + "failed to upload '{}' to '{}'".format(from_info, to_info) ) def download( @@ -283,28 +302,26 @@ class RemoteLOCAL(RemoteBASE): if to_info.scheme != "local": raise NotImplementedError - logger.debug( - "Downloading '{}' to '{}'".format(from_info.path, to_info.path) - ) + logger.debug("Downloading '{}' to '{}'".format(from_info, to_info)) if not name: - name = os.path.basename(to_info.path) + name = to_info.name - makedirs(os.path.dirname(to_info.path), exist_ok=True) - tmp_file = tmp_fname(to_info.path) + makedirs(fspath_py35(to_info.parent), exist_ok=True) + tmp_file = tmp_fname(to_info) try: copyfile( - from_info.path, + fspath_py35(from_info), tmp_file, no_progress_bar=no_progress_bar, name=name, ) - move(tmp_file, to_info.path) + move(tmp_file, fspath_py35(to_info)) except Exception: logger.exception( "failed to download '{}' to '{}'".format( - from_info.path, to_info.path + from_info, to_info ) ) @@ -340,7 +357,9 @@ class RemoteLOCAL(RemoteBASE): show_checksums=False, download=False, ): - logger.info("Preparing to collect status from {}".format(remote.url)) + logger.info( + "Preparing to collect status from {}".format(remote.path_info) + ) title = "Collecting information" ret = {} @@ -351,7 +370,7 @@ class RemoteLOCAL(RemoteBASE): progress.update_target(title, 10, 100) ret = self._group(checksum_infos, show_checksums=show_checksums) - md5s = list(ret.keys()) + md5s = list(ret) progress.update_target(title, 30, 100) @@ -437,7 +456,7 @@ class RemoteLOCAL(RemoteBASE): msg.format( "download" if download else "upload", "from" if download else "to", - remote.url, + remote.path_info, ) ) @@ -552,7 +571,7 @@ class RemoteLOCAL(RemoteBASE): @staticmethod def unprotect(path_info): - path = path_info.path + path = path_info.fspath if not os.path.exists(path): raise DvcException( "can't unprotect non-existing data '{}'".format(path) @@ -565,4 +584,57 @@ class RemoteLOCAL(RemoteBASE): @staticmethod def protect(path_info): - os.chmod(path_info.path, stat.S_IREAD | stat.S_IRGRP | stat.S_IROTH) + os.chmod( + fspath_py35(path_info), stat.S_IREAD | stat.S_IRGRP | stat.S_IROTH + ) + + def _get_unpacked_dir_path_info(self, checksum): + info = self.checksum_to_path_info(checksum) + return info.with_name(info.name + self.UNPACKED_DIR_SUFFIX) + + def _path_info_changed(self, path_info): + if self.exists(path_info) and self.state.get(path_info): + return False + return True + + def _update_unpacked_dir(self, checksum): + unpacked_dir_info = self._get_unpacked_dir_path_info(checksum) + + if not self._path_info_changed(unpacked_dir_info): + return + + self.remove(unpacked_dir_info) + + try: + dir_info = self.get_dir_cache(checksum) + self._create_unpacked_dir(checksum, dir_info, unpacked_dir_info) + except DvcException: + logger.warning("Could not create '{}'".format(unpacked_dir_info)) + + self.remove(unpacked_dir_info) + + def _create_unpacked_dir(self, checksum, dir_info, unpacked_dir_info): + self.makedirs(unpacked_dir_info) + + for entry in progress(dir_info, name="Created unpacked dir"): + entry_cache_info = self.checksum_to_path_info( + entry[self.PARAM_CHECKSUM] + ) + relpath = entry[self.PARAM_RELPATH] + self.link( + entry_cache_info, unpacked_dir_info / relpath, "hardlink" + ) + + self.state.save(unpacked_dir_info, checksum) + + def _changed_unpacked_dir(self, checksum): + status_unpacked_dir_info = self._get_unpacked_dir_path_info(checksum) + + return not self.state.get(status_unpacked_dir_info) + + def _get_unpacked_dir_names(self, checksums): + unpacked = set() + for c in checksums: + if self.is_dir_checksum(c): + unpacked.add(c + self.UNPACKED_DIR_SUFFIX) + return unpacked diff --git a/dvc/remote/oss.py b/dvc/remote/oss.py index 56879501d..4dc1818e8 100644 --- a/dvc/remote/oss.py +++ b/dvc/remote/oss.py @@ -5,7 +5,6 @@ import os import logging from dvc.scheme import Schemes -from dvc.path.oss import PathOSS try: import oss2 @@ -13,11 +12,12 @@ except ImportError: oss2 = None from dvc.utils import tmp_fname, move -from dvc.utils.compat import urlparse, makedirs +from dvc.utils.compat import makedirs, fspath_py35 from dvc.progress import progress from dvc.config import Config from dvc.remote.base import RemoteBASE from dvc.remote.azure import Callback +from dvc.path_info import CloudURLInfo logger = logging.getLogger(__name__) @@ -43,18 +43,17 @@ class RemoteOSS(RemoteBASE): """ scheme = Schemes.OSS - REGEX = r"^oss://(?P.*)?$" + path_cls = CloudURLInfo REQUIRES = {"oss2": oss2} PARAM_CHECKSUM = "etag" + SUPPORTS_NO_TRAVERSE = False COPY_POLL_SECONDS = 5 def __init__(self, repo, config): super(RemoteOSS, self).__init__(repo, config) - self.url = config.get(Config.SECTION_REMOTE_URL) - parsed = urlparse(self.url) - self.bucket = parsed.netloc - self.prefix = parsed.path.lstrip("/") + url = config.get(Config.SECTION_REMOTE_URL) + self.path_info = self.path_cls(url) if url else None self.endpoint = config.get(Config.SECTION_OSS_ENDPOINT) or os.getenv( "OSS_ENDPOINT" @@ -73,17 +72,17 @@ class RemoteOSS(RemoteBASE): ) self._bucket = None - self.path_info = PathOSS(bucket=self.bucket) @property def oss_service(self): if self._bucket is None: - logger.debug("URL {}".format(self.url)) + logger.debug("URL {}".format(self.path_info)) logger.debug("key id {}".format(self.key_id)) logger.debug("key secret {}".format(self.key_secret)) auth = oss2.Auth(self.key_id, self.key_secret) - logger.debug("bucket name {}".format(self.bucket)) - self._bucket = oss2.Bucket(auth, self.endpoint, self.bucket) + self._bucket = oss2.Bucket( + auth, self.endpoint, self.path_info.bucket + ) try: # verify that bucket exists self._bucket.get_bucket_info() except oss2.exceptions.NoSuchBucket: @@ -99,10 +98,7 @@ class RemoteOSS(RemoteBASE): if path_info.scheme != self.scheme: raise NotImplementedError - logger.debug( - "Removing oss://{}/{}".format(path_info.bucket, path_info.path) - ) - + logger.debug("Removing oss://{}".format(path_info)) self.oss_service.delete_object(path_info.path) def _list_paths(self, prefix): @@ -110,7 +106,7 @@ class RemoteOSS(RemoteBASE): yield blob.key def list_cache_paths(self): - return self._list_paths(self.prefix) + return self._list_paths(self.path_info.path) def upload(self, from_infos, to_infos, names=None, no_progress_bar=False): names = self._verify_path_args(to_infos, from_infos, names) @@ -122,26 +118,19 @@ class RemoteOSS(RemoteBASE): if from_info.scheme != "local": raise NotImplementedError - bucket = to_info.bucket - path = to_info.path - - logger.debug( - "Uploading '{}' to 'oss://{}/{}'".format( - from_info.path, bucket, path - ) - ) + logger.debug("Uploading '{}' to '{}'".format(from_info, to_info)) if not name: - name = os.path.basename(from_info.path) + name = from_info.name cb = None if no_progress_bar else Callback(name) try: self.oss_service.put_object_from_file( - path, from_info.path, progress_callback=cb + to_info.path, from_info.fspath, progress_callback=cb ) except Exception: - msg = "failed to upload '{}'".format(from_info.path) + msg = "failed to upload '{}'".format(from_info) logger.warning(msg) else: progress.finish_target(name) @@ -161,32 +150,23 @@ class RemoteOSS(RemoteBASE): if to_info.scheme != "local": raise NotImplementedError - bucket = from_info.bucket - path = from_info.path + logger.debug("Downloading '{}' to '{}'".format(from_info, to_info)) - logger.debug( - "Downloading 'oss://{}/{}' to '{}'".format( - bucket, path, to_info.path - ) - ) - - tmp_file = tmp_fname(to_info.path) + tmp_file = tmp_fname(to_info) if not name: - name = os.path.basename(to_info.path) + name = to_info.name cb = None if no_progress_bar else Callback(name) - makedirs(os.path.dirname(to_info.path), exist_ok=True) - + makedirs(fspath_py35(to_info.parent), exist_ok=True) try: self.oss_service.get_object_to_file( - path, tmp_file, progress_callback=cb + from_info.path, tmp_file, progress_callback=cb ) except Exception: - msg = "failed to download 'oss://{}/{}'".format(bucket, path) - logger.warning(msg) + logger.warning("failed to download '{}'".format(from_info)) else: - move(tmp_file, to_info.path) - + move(tmp_file, fspath_py35(to_info)) + finally: if not no_progress_bar: progress.finish_target(name) diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 900f8b7bb..e8087a8a9 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -3,9 +3,9 @@ from __future__ import unicode_literals import os import threading import logging +import itertools from dvc.scheme import Schemes -from dvc.path.s3 import PathS3 try: import boto3 @@ -13,11 +13,12 @@ except ImportError: boto3 = None from dvc.utils import tmp_fname, move -from dvc.utils.compat import urlparse, makedirs +from dvc.utils.compat import makedirs, fspath_py35 from dvc.progress import progress from dvc.config import Config from dvc.remote.base import RemoteBASE from dvc.exceptions import DvcException, ETagMismatchError +from dvc.path_info import CloudURLInfo logger = logging.getLogger(__name__) @@ -37,7 +38,7 @@ class Callback(object): class RemoteS3(RemoteBASE): scheme = Schemes.S3 - REGEX = r"^s3://(?P.*)$" + path_cls = CloudURLInfo REQUIRES = {"boto3": boto3} PARAM_CHECKSUM = "etag" @@ -48,7 +49,8 @@ class RemoteS3(RemoteBASE): config.get(Config.SECTION_AWS_STORAGEPATH, "").lstrip("/") ) - self.url = config.get(Config.SECTION_REMOTE_URL, storagepath) + url = config.get(Config.SECTION_REMOTE_URL, storagepath) + self.path_info = self.path_cls(url) self.region = os.environ.get("AWS_DEFAULT_REGION") or config.get( Config.SECTION_AWS_REGION @@ -74,12 +76,6 @@ class RemoteS3(RemoteBASE): if shared_creds: os.environ.setdefault("AWS_SHARED_CREDENTIALS_FILE", shared_creds) - parsed = urlparse(self.url) - self.bucket = parsed.netloc - self.prefix = parsed.path.lstrip("/") - - self.path_info = PathS3(bucket=self.bucket) - @staticmethod def compat_config(config): ret = config.copy() @@ -117,16 +113,19 @@ class RemoteS3(RemoteBASE): ) return obj - def _copy_multipart(self, s3, from_info, to_info, size, n_parts): + @classmethod + def _copy_multipart( + cls, s3, from_info, to_info, size, n_parts, extra_args + ): mpu = s3.create_multipart_upload( - Bucket=to_info.bucket, Key=to_info.path, **self.extra_args + Bucket=to_info.bucket, Key=to_info.path, **extra_args ) mpu_id = mpu["UploadId"] parts = [] byte_position = 0 for i in range(1, n_parts + 1): - obj = self.get_head_object( + obj = cls.get_head_object( s3, from_info.bucket, from_info.path, PartNumber=i ) part_size = obj["ContentLength"] @@ -158,7 +157,8 @@ class RemoteS3(RemoteBASE): MultipartUpload={"Parts": parts}, ) - def _copy(self, s3, from_info, to_info): + @classmethod + def _copy(cls, s3, from_info, to_info, extra_args): # NOTE: object's etag depends on the way it was uploaded to s3 or the # way it was copied within the s3. More specifically, it depends on # the chunk size that was used to transfer it, which would affect @@ -179,41 +179,38 @@ class RemoteS3(RemoteBASE): # preserve etag, we need to transfer each part separately, so the # object is transfered in the same chunks as it was originally. - obj = self.get_head_object(s3, from_info.bucket, from_info.path) + obj = cls.get_head_object(s3, from_info.bucket, from_info.path) etag = obj["ETag"].strip('"') size = obj["ContentLength"] _, _, parts_suffix = etag.partition("-") if parts_suffix: n_parts = int(parts_suffix) - self._copy_multipart(s3, from_info, to_info, size, n_parts) + cls._copy_multipart( + s3, from_info, to_info, size, n_parts, extra_args=extra_args + ) else: source = {"Bucket": from_info.bucket, "Key": from_info.path} - s3.copy( - source, to_info.bucket, to_info.path, ExtraArgs=self.extra_args - ) + s3.copy(source, to_info.bucket, to_info.path, ExtraArgs=extra_args) - cached_etag = self.get_etag(s3, to_info.bucket, to_info.path) + cached_etag = cls.get_etag(s3, to_info.bucket, to_info.path) if etag != cached_etag: raise ETagMismatchError(etag, cached_etag) def copy(self, from_info, to_info, s3=None): s3 = s3 if s3 else self.s3 - self._copy(s3, from_info, to_info) + self._copy(s3, from_info, to_info, self.extra_args) def remove(self, path_info): if path_info.scheme != "s3": raise NotImplementedError - logger.debug( - "Removing s3://{}/{}".format(path_info.bucket, path_info.path) - ) - + logger.debug("Removing {}".format(path_info)) self.s3.delete_object(Bucket=path_info.bucket, Key=path_info.path) - def _list_paths(self, bucket, prefix): + def _list_paths(self, bucket, prefix, s3=None): """ Read config for list object api, paginate through list objects.""" - s3 = self.s3 + s3 = s3 or self.s3 kwargs = {"Bucket": bucket, "Prefix": prefix} if self.list_objects: list_objects_api = "list_objects" @@ -228,14 +225,30 @@ class RemoteS3(RemoteBASE): yield item["Key"] def list_cache_paths(self): - return self._list_paths(self.bucket, self.prefix) + return self._list_paths(self.path_info.bucket, self.path_info.path) - def exists(self, path_info): - assert not isinstance(path_info, list) - assert path_info.scheme == "s3" + def exists(self, path_infos): + single_path = False - paths = self._list_paths(path_info.bucket, path_info.path) - return any(path_info.path == path for path in paths) + if not isinstance(path_infos, list): + single_path = True + path_infos = [path_infos] + + s3 = self.s3 + + paths = itertools.chain.from_iterable( + self._list_paths(path_info.bucket, path_info.path, s3) + for path_info in path_infos + ) + + paths = set(paths) + + results = [path_info.path in paths for path_info in path_infos] + + if single_path and results: + return all(results) + + return results def upload(self, from_infos, to_infos, names=None, no_progress_bar=False): names = self._verify_path_args(to_infos, from_infos, names) @@ -249,28 +262,24 @@ class RemoteS3(RemoteBASE): if from_info.scheme != "local": raise NotImplementedError - logger.debug( - "Uploading '{}' to '{}/{}'".format( - from_info.path, to_info.bucket, to_info.path - ) - ) + logger.debug("Uploading '{}' to '{}'".format(from_info, to_info)) if not name: - name = os.path.basename(from_info.path) + name = from_info.name - total = os.path.getsize(from_info.path) + total = os.path.getsize(fspath_py35(from_info)) cb = None if no_progress_bar else Callback(name, total) try: s3.upload_file( - from_info.path, + from_info.fspath, to_info.bucket, to_info.path, Callback=cb, ExtraArgs=self.extra_args, ) except Exception: - msg = "failed to upload '{}'".format(from_info.path) + msg = "failed to upload '{}'".format(from_info) logger.exception(msg) continue @@ -299,17 +308,14 @@ class RemoteS3(RemoteBASE): if to_info.scheme != "local": raise NotImplementedError - msg = "Downloading '{}/{}' to '{}'".format( - from_info.bucket, from_info.path, to_info.path - ) + msg = "Downloading '{}' to '{}'".format(from_info, to_info) logger.debug(msg) - tmp_file = tmp_fname(to_info.path) + tmp_file = tmp_fname(to_info) if not name: - name = os.path.basename(to_info.path) - - makedirs(os.path.dirname(to_info.path), exist_ok=True) + name = to_info.name + makedirs(fspath_py35(to_info.parent), exist_ok=True) try: if no_progress_bar: cb = None @@ -323,13 +329,11 @@ class RemoteS3(RemoteBASE): from_info.bucket, from_info.path, tmp_file, Callback=cb ) except Exception: - msg = "failed to download '{}/{}'".format( - from_info.bucket, from_info.path - ) + msg = "failed to download '{}'".format(from_info) logger.exception(msg) continue - move(tmp_file, to_info.path) + move(tmp_file, fspath_py35(to_info)) if not no_progress_bar: progress.finish_target(name) diff --git a/dvc/remote/ssh/__init__.py b/dvc/remote/ssh/__init__.py index a8cf29c38..a067db341 100644 --- a/dvc/remote/ssh/__init__.py +++ b/dvc/remote/ssh/__init__.py @@ -4,8 +4,7 @@ import os import getpass import logging -from dvc.path import Schemes -from dvc.path.ssh import PathSSH +from dvc.scheme import Schemes try: import paramiko @@ -15,7 +14,7 @@ except ImportError: import dvc.prompt as prompt from dvc.remote.ssh.connection import SSHConnection from dvc.config import Config -from dvc.utils.compat import urlparse +from dvc.utils.compat import urlparse, StringIO from dvc.remote.base import RemoteBASE @@ -24,11 +23,6 @@ logger = logging.getLogger(__name__) class RemoteSSH(RemoteBASE): scheme = Schemes.SSH - - # NOTE: we support both URL-like (ssh://[user@]host.xz[:port]/path) and - # SCP-like (ssh://[user@]host.xz:/absolute/path) urls. - REGEX = r"^ssh://.*$" - REQUIRES = {"paramiko": paramiko} JOBS = 4 @@ -38,27 +32,36 @@ class RemoteSSH(RemoteBASE): def __init__(self, repo, config): super(RemoteSSH, self).__init__(repo, config) - self.url = config.get(Config.SECTION_REMOTE_URL, "ssh://") - parsed = urlparse(self.url) - self.host = parsed.hostname + url = config.get(Config.SECTION_REMOTE_URL) + if url: + parsed = urlparse(url) + user_ssh_config = self._load_user_ssh_config(parsed.hostname) - user_ssh_config = self._load_user_ssh_config(self.host) + host = user_ssh_config.get("hostname", parsed.hostname) + user = ( + config.get(Config.SECTION_REMOTE_USER) + or parsed.username + or user_ssh_config.get("user") + or getpass.getuser() + ) + port = ( + config.get(Config.SECTION_REMOTE_PORT) + or parsed.port + or self._try_get_ssh_config_port(user_ssh_config) + or self.DEFAULT_PORT + ) + self.path_info = self.path_cls.from_parts( + scheme=self.scheme, + host=host, + user=user, + port=port, + path=parsed.path, + ) + else: + self.path_info = None + user_ssh_config = {} - self.host = user_ssh_config.get("hostname", self.host) - self.user = ( - config.get(Config.SECTION_REMOTE_USER) - or parsed.username - or user_ssh_config.get("user") - or getpass.getuser() - ) - self.prefix = parsed.path or "/" - self.port = ( - config.get(Config.SECTION_REMOTE_PORT) - or parsed.port - or self._try_get_ssh_config_port(user_ssh_config) - or self.DEFAULT_PORT - ) self.keyfile = config.get( Config.SECTION_REMOTE_KEY_FILE ) or self._try_get_ssh_config_keyfile(user_ssh_config) @@ -68,10 +71,6 @@ class RemoteSSH(RemoteBASE): Config.SECTION_REMOTE_ASK_PASSWORD, False ) - self.path_info = PathSSH( - host=self.host, user=self.user, port=self.port - ) - @staticmethod def ssh_config_filename(): return os.path.expanduser(os.path.join("~", ".ssh", "config")) @@ -81,10 +80,12 @@ class RemoteSSH(RemoteBASE): user_config_file = RemoteSSH.ssh_config_filename() user_ssh_config = dict() if hostname and os.path.exists(user_config_file): + ssh_config = paramiko.SSHConfig() with open(user_config_file) as f: - ssh_config = paramiko.SSHConfig() - ssh_config.parse(f) - user_ssh_config = ssh_config.lookup(hostname) + # For whatever reason parsing directly from f is unreliable + f_copy = StringIO(f.read()) + ssh_config.parse(f_copy) + user_ssh_config = ssh_config.lookup(hostname) return user_ssh_config @staticmethod @@ -128,12 +129,20 @@ class RemoteSSH(RemoteBASE): password=self.password, ) - def exists(self, path_info): - assert not isinstance(path_info, list) - assert path_info.scheme == self.scheme + def exists(self, path_infos): + single_path = False - with self.ssh(path_info) as ssh: - return ssh.exists(path_info.path) + if not isinstance(path_infos, list): + single_path = True + path_infos = [path_infos] + + with self.ssh(path_infos[0]) as ssh: + results = [ssh.exists(path_info.path) for path_info in path_infos] + + if single_path and results: + return all(results) + + return results def get_file_checksum(self, path_info): if path_info.scheme != self.scheme: @@ -190,24 +199,22 @@ class RemoteSSH(RemoteBASE): raise NotImplementedError logger.debug( - "Downloading '{host}/{path}' to '{dest}'".format( - host=from_info.host, path=from_info.path, dest=to_info.path + "Downloading '{src}' to '{dest}'".format( + src=from_info, dest=to_info ) ) try: ssh.download( from_info.path, - to_info.path, + to_info.fspath, progress_title=name, no_progress_bar=no_progress_bar, ) except Exception: logger.exception( - "failed to download '{host}/{path}' to '{dest}'".format( - host=from_info.host, - path=from_info.path, - dest=to_info.path, + "failed to download '{src}' to '{dest}'".format( + src=from_info, dest=to_info ) ) continue @@ -227,7 +234,7 @@ class RemoteSSH(RemoteBASE): try: ssh.upload( - from_info.path, + from_info.fspath, to_info.path, progress_title=name, no_progress_bar=no_progress_bar, @@ -241,7 +248,7 @@ class RemoteSSH(RemoteBASE): def list_cache_paths(self): with self.ssh(self.path_info) as ssh: - return list(ssh.walk_files(self.prefix)) + return list(ssh.walk_files(self.path_info.path)) def walk(self, path_info): with self.ssh(path_info) as ssh: diff --git a/dvc/repo/__init__.py b/dvc/repo/__init__.py index 884852947..0182b8ac8 100644 --- a/dvc/repo/__init__.py +++ b/dvc/repo/__init__.py @@ -13,7 +13,7 @@ from dvc.exceptions import ( TargetNotDirectoryError, ) from dvc.ignore import DvcIgnoreFileHandler -from dvc.path.local import PathLOCAL +from dvc.path_info import PathInfo from dvc.utils.compat import open as _open logger = logging.getLogger(__name__) @@ -118,8 +118,7 @@ class Repo(object): return Repo(root_dir) def unprotect(self, target): - path_info = PathLOCAL(path=target) - return self.cache.local.unprotect(path_info) + return self.cache.local.unprotect(PathInfo(target)) def _ignore(self): flist = [ @@ -397,24 +396,24 @@ class Repo(object): for out in stage.outs: existing = [] for o in outs: - if o.path == out.path: + if o.path_info == out.path_info: existing.append(o.stage) - in_o_dir = out.path.startswith(o.path + o.sep) - in_out_dir = o.path.startswith(out.path + out.sep) + in_o_dir = out.path_info.isin(o.path_info) + in_out_dir = o.path_info.isin(out.path_info) if in_o_dir or in_out_dir: raise OverlappingOutputPathsError(o, out) if existing: stages = [stage.relpath, existing[0].relpath] - raise OutputDuplicationError(out.path, stages) + raise OutputDuplicationError(str(out), stages) outs.append(out) for stage in stages: - path_dir = os.path.dirname(stage.path) + os.sep + stage_path_info = PathInfo(stage.path) for out in outs: - if path_dir.startswith(out.path + os.sep): + if stage_path_info.isin(out.path_info): raise StagePathAsOutputError(stage.wdir, stage.relpath) for stage in stages: @@ -426,9 +425,9 @@ class Repo(object): for dep in stage.deps: for out in outs: if ( - out.path != dep.path - and not dep.path.startswith(out.path + out.sep) - and not out.path.startswith(dep.path + dep.sep) + out.path_info != dep.path_info + and not dep.path_info.isin(out.path_info) + and not out.path_info.isin(dep.path_info) ): continue @@ -483,7 +482,8 @@ class Repo(object): continue stage = Stage.load(self, path) for out in stage.outs: - outs.append(out.path + out.sep) + if out.scheme == "local": + outs.append(out.fspath + out.sep) stages.append(stage) def filter_dirs(dname, root=root): @@ -522,10 +522,10 @@ class Repo(object): is_dir = self.tree.isdir(abs_path) def func(out): - if out.path == abs_path: + if out.scheme == "local" and out.fspath == abs_path: return True - if is_dir and recursive and out.path.startswith(abs_path + os.sep): + if is_dir and recursive and out.path_info.isin(abs_path): return True return False @@ -550,5 +550,5 @@ class Repo(object): cache_info = self._collect_used_cache(out, remote=remote) self.cloud.pull(cache_info, remote=remote) - cache_filename = self.cache.local.checksum_to_path(out.checksum) - return _open(cache_filename, mode=mode, encoding=encoding) + cache_file = self.cache.local.checksum_to_path_info(out.checksum) + return _open(cache_file.fspath, mode=mode, encoding=encoding) diff --git a/dvc/repo/checkout.py b/dvc/repo/checkout.py index 2e1f3dd9f..2d743e41d 100644 --- a/dvc/repo/checkout.py +++ b/dvc/repo/checkout.py @@ -10,10 +10,12 @@ logger = logging.getLogger(__name__) def _cleanup_unused_links(self, all_stages): - used = [] - for stage in all_stages: - for out in stage.outs: - used.append(out.path) + used = [ + out.fspath + for stage in all_stages + for out in stage.outs + if out.scheme == "local" + ] self.state.remove_unused_links(used) diff --git a/dvc/repo/metrics/show.py b/dvc/repo/metrics/show.py index 2bec09bc8..6ce9f8e71 100644 --- a/dvc/repo/metrics/show.py +++ b/dvc/repo/metrics/show.py @@ -134,7 +134,7 @@ def _format_output(content, typ): return content -def _read_metric(fd, typ=None, xpath=None, rel_path=None, branch=None): +def _read_metric(fd, typ=None, xpath=None, fname=None, branch=None): typ = typ.lower().strip() if typ else typ try: if xpath: @@ -146,7 +146,7 @@ def _read_metric(fd, typ=None, xpath=None, rel_path=None, branch=None): except Exception: logger.exception( "unable to read metric in '{}' in branch '{}'".format( - rel_path, branch + fname, branch ) ) return None @@ -220,28 +220,24 @@ def _read_metrics(repo, metrics, branch): for out, typ, xpath in metrics: assert out.scheme == "local" if not typ: - typ = os.path.splitext(out.path.lower())[1].replace(".", "") + typ = os.path.splitext(out.fspath.lower())[1].replace(".", "") if out.use_cache: open_fun = open path = repo.cache.local.get(out.checksum) else: open_fun = repo.tree.open - path = out.path + path = out.fspath try: with open_fun(path) as fd: metric = _read_metric( - fd, - typ=typ, - xpath=xpath, - rel_path=out.rel_path, - branch=branch, + fd, typ=typ, xpath=xpath, fname=str(out), branch=branch ) except IOError as e: if e.errno == errno.ENOENT: logger.warning( NO_METRICS_FILE_AT_REFERENCE_WARNING.format( - out.rel_path, branch + out.path_info, branch ) ) metric = None @@ -251,7 +247,7 @@ def _read_metrics(repo, metrics, branch): if not metric: continue - res[out.rel_path] = metric + res[str(out)] = metric return res diff --git a/dvc/repo/move.py b/dvc/repo/move.py index 6b9e812ed..7b5dd6c61 100644 --- a/dvc/repo/move.py +++ b/dvc/repo/move.py @@ -33,10 +33,11 @@ def move(self, from_path, to_path): from dvc.stage import Stage from_out = Output.loads_from(Stage(self), [from_path])[0] + assert from_out.scheme == "local" to_path = _expand_target_path(from_path, to_path) - outs = self.find_outs_by_path(from_out.path) + outs = self.find_outs_by_path(from_out.fspath) assert len(outs) == 1 out = outs[0] stage = out.stage @@ -45,7 +46,7 @@ def move(self, from_path, to_path): raise MoveNotDataSourceError(stage.relpath) stage_name = os.path.splitext(os.path.basename(stage.path))[0] - from_name = os.path.basename(from_out.path) + from_name = os.path.basename(from_out.fspath) if stage_name == from_name: os.unlink(stage.path) diff --git a/dvc/repo/pkg/install.py b/dvc/repo/pkg/install.py index d5f50669f..671f2e9d9 100644 --- a/dvc/repo/pkg/install.py +++ b/dvc/repo/pkg/install.py @@ -118,7 +118,7 @@ class GitPackage(Package): ) for out in stage.outs: - parent_repo.scm.ignore(out.path, in_curr_dir=True) + parent_repo.scm.ignore(out.fspath, in_curr_dir=True) stage.dump() diff --git a/dvc/scm/git/temp_git_repo.py b/dvc/scm/git/temp_git_repo.py index 9793a08cf..c40a5f36b 100644 --- a/dvc/scm/git/temp_git_repo.py +++ b/dvc/scm/git/temp_git_repo.py @@ -116,7 +116,7 @@ class TempGitRepo(object): if not self.is_state_set: raise TempRepoException(self, "cannot persist") - tmp_repo_cache = self.repo.cache.local.url + tmp_repo_cache = self.repo.cache.local.path_info.fspath for prefix in os.listdir(tmp_repo_cache): if len(prefix) != 2: @@ -135,7 +135,9 @@ class TempGitRepo(object): obj_name = os.path.join(tmp_repo_cache, prefix) for suffix in os.listdir(obj_name): src_path = os.path.join(tmp_repo_cache, prefix, suffix) - pre = os.path.join(parent_repo.cache.local.url, prefix) + pre = os.path.join( + parent_repo.cache.local.path_info.fspath, prefix + ) if not os.path.exists(pre): os.mkdir(pre) dest = os.path.join(pre, suffix) diff --git a/dvc/stage.py b/dvc/stage.py index 3fece0483..118ee6e18 100644 --- a/dvc/stage.py +++ b/dvc/stage.py @@ -8,6 +8,7 @@ import os import subprocess import logging +from dvc.utils.compat import pathlib from dvc.utils.fs import contains_symlink_up_to from schema import Schema, SchemaError, Optional, Or, And @@ -322,32 +323,20 @@ class Stage(object): raise StageFileFormatError(fname, exc) @classmethod - def _stage_fname(cls, fname, outs, add): - if fname: - return fname - + def _stage_fname(cls, outs, add): if not outs: return cls.STAGE_FILE out = outs[0] - path_handler = out.remote.ospath + fname = out.path_info.name + cls.STAGE_FILE_SUFFIX - fname = path_handler.basename(out.path) + cls.STAGE_FILE_SUFFIX - - fname = Stage._expand_to_path_on_add_local( - add, fname, out, path_handler - ) - - return fname - - @staticmethod - def _expand_to_path_on_add_local(add, fname, out, path_handler): if ( add and out.is_in_repo - and not contains_symlink_up_to(out.path, out.repo.root_dir) + and not contains_symlink_up_to(out.fspath, out.repo.root_dir) ): - fname = path_handler.join(path_handler.dirname(out.path), fname) + fname = out.path_info.with_name(fname).fspath + return fname @staticmethod @@ -475,7 +464,8 @@ class Stage(object): stage._check_circular_dependency() stage._check_duplicated_arguments() - fname = Stage._stage_fname(fname, stage.outs, add=add) + if not fname: + fname = Stage._stage_fname(stage.outs, add=add) wdir = os.path.abspath(wdir) if cwd is not None: @@ -619,16 +609,13 @@ class Stage(object): return stage def dumpd(self): - from dvc.remote.base import RemoteBASE - + rel_wdir = os.path.relpath(self.wdir, os.path.dirname(self.path)) return { key: value for key, value in { Stage.PARAM_MD5: self.md5, Stage.PARAM_CMD: self.cmd, - Stage.PARAM_WDIR: RemoteBASE.to_posixpath( - os.path.relpath(self.wdir, os.path.dirname(self.path)) - ), + Stage.PARAM_WDIR: pathlib.PurePath(rel_wdir).as_posix(), Stage.PARAM_LOCKED: self.locked, Stage.PARAM_DEPS: [d.dumpd() for d in self.deps], Stage.PARAM_OUTS: [o.dumpd() for o in self.outs], @@ -695,11 +682,11 @@ class Stage(object): @staticmethod def _changed_entries(entries): - ret = [] - for entry in entries: - if entry.checksum and entry.changed_checksum(): - ret.append(entry.rel_path) - return ret + return [ + str(entry) + for entry in entries + if entry.checksum and entry.changed_checksum() + ] def check_can_commit(self, force): changed_deps = self._changed_entries(self.deps) @@ -751,22 +738,22 @@ class Stage(object): def _check_circular_dependency(self): from dvc.exceptions import CircularDependencyError - circular_dependencies = set(d.path for d in self.deps) & set( - o.path for o in self.outs + circular_dependencies = set(d.path_info for d in self.deps) & set( + o.path_info for o in self.outs ) if circular_dependencies: - raise CircularDependencyError(circular_dependencies.pop()) + raise CircularDependencyError(str(circular_dependencies.pop())) def _check_duplicated_arguments(self): from dvc.exceptions import ArgumentDuplicationError from collections import Counter - path_counts = Counter(edge.path for edge in self.deps + self.outs) + path_counts = Counter(edge.path_info for edge in self.deps + self.outs) for path, occurrence in path_counts.items(): if occurrence > 1: - raise ArgumentDuplicationError(path) + raise ArgumentDuplicationError(str(path)) def _run(self): self._check_missing_deps() @@ -801,7 +788,7 @@ class Stage(object): elif self.is_import: logger.info( "Importing '{dep}' -> '{out}'".format( - dep=self.deps[0].path, out=self.outs[0].path + dep=self.deps[0], out=self.outs[0] ) ) if not dry: @@ -836,12 +823,7 @@ class Stage(object): self.commit() def check_missing_outputs(self): - paths = [ - out.path if out.scheme != "local" else out.rel_path - for out in self.outs - if not out.exists - ] - + paths = [str(out) for out in self.outs if not out.exists] if paths: raise MissingDataSource(paths) diff --git a/dvc/state.py b/dvc/state.py index 08dbab63c..ed9656717 100644 --- a/dvc/state.py +++ b/dvc/state.py @@ -10,6 +10,7 @@ from dvc.config import Config from dvc.utils import remove, current_timestamp from dvc.exceptions import DvcException from dvc.utils.fs import get_mtime_and_size, get_inode +from dvc.utils.compat import fspath logger = logging.getLogger(__name__) @@ -352,7 +353,7 @@ class State(StateBase): # pylint: disable=too-many-instance-attributes assert path_info.scheme == "local" assert checksum is not None - path = path_info.path + path = fspath(path_info) assert os.path.exists(path) actual_mtime, actual_size = get_mtime_and_size(path) @@ -381,7 +382,7 @@ class State(StateBase): # pylint: disable=too-many-instance-attributes doesn't exist in the state database. """ assert path_info.scheme == "local" - path = path_info.path + path = fspath(path_info) if not os.path.exists(path): return None @@ -408,7 +409,7 @@ class State(StateBase): # pylint: disable=too-many-instance-attributes path_info (dict): path info to add to the list of links. """ assert path_info.scheme == "local" - path = path_info.path + path = fspath(path_info) if not os.path.exists(path): return diff --git a/dvc/utils/__init__.py b/dvc/utils/__init__.py index ee31900e2..85ee5b68a 100644 --- a/dvc/utils/__init__.py +++ b/dvc/utils/__init__.py @@ -3,6 +3,7 @@ from __future__ import unicode_literals from dvc.utils.compat import str, builtin_str, open, cast_bytes_py2, StringIO +from dvc.utils.compat import fspath import os import sys @@ -234,7 +235,7 @@ def tmp_fname(fname): """ Temporary name for a partial download """ from uuid import uuid4 - return fname + "." + str(uuid4()) + ".tmp" + return fspath(fname) + "." + str(uuid4()) + ".tmp" def current_timestamp(): diff --git a/dvc/utils/collections.py b/dvc/utils/collections.py index e1367d10a..2eb47ba76 100644 --- a/dvc/utils/collections.py +++ b/dvc/utils/collections.py @@ -2,11 +2,6 @@ from __future__ import absolute_import, unicode_literals from dvc.utils.compat import Mapping -# just simple check for Nones and emtpy strings -def compact(args): - return list(filter(bool, args)) - - def apply_diff(src, dest): """Recursively apply changes from src to dest. diff --git a/dvc/utils/compat.py b/dvc/utils/compat.py index 2c3ec53e0..0abb08c4a 100644 --- a/dvc/utils/compat.py +++ b/dvc/utils/compat.py @@ -96,7 +96,7 @@ if is_py2: from SimpleHTTPServer import SimpleHTTPRequestHandler # noqa: F401 import ConfigParser # noqa: F401 from io import open # noqa: F401 - from pathlib2 import Path # noqa: F401 + import pathlib2 as pathlib # noqa: F401 from collections import Mapping # noqa: F401 builtin_str = str # noqa: F821 @@ -129,7 +129,7 @@ if is_py2: elif is_py3: - from pathlib import Path # noqa: F401 + import pathlib # noqa: F401 from os import makedirs # noqa: F401 from urllib.parse import ( # noqa: F401 urlparse, # noqa: F401 @@ -155,3 +155,45 @@ elif is_py3: open = open # noqa: F821 cast_bytes_py2 = no_code range = range # noqa: F821 + + +# Backport os.fspath() from Python 3.6 +try: + from os import fspath # noqa: F821 + + fspath_py35 = lambda s: s # noqa: E731 +except ImportError: + + def fspath(path): + """Return the path representation of a path-like object. + + If str or bytes is passed in, it is returned unchanged. Otherwise the + os.PathLike interface is used to get the path representation. If the + path representation is not str or bytes, TypeError is raised. If the + provided path is not str, bytes, or os.PathLike, TypeError is raised. + """ + if isinstance(path, (str, bytes)): + return path + + # Work from the object's type to match method resolution of other magic + # methods. + path_type = type(path) + try: + path_repr = path_type.__fspath__(path) + except AttributeError: + if hasattr(path_type, "__fspath__"): + raise + else: + raise TypeError( + "expected str, bytes or os.PathLike object, " + "not " + path_type.__name__ + ) + if isinstance(path_repr, (str, bytes)): + return path_repr + else: + raise TypeError( + "expected {}.__fspath__() to return str or bytes, " + "not {}".format(path_type.__name__, type(path_repr).__name__) + ) + + fspath_py35 = fspath diff --git a/setup.py b/setup.py index d8a0c4972..dc413d489 100644 --- a/setup.py +++ b/setup.py @@ -61,6 +61,7 @@ install_requires = [ "dulwich>=0.19.11", "ruamel.yaml>=0.15.91", "psutil==5.6.2", + "funcy>=1.12", ] # Extra dependencies for remote integrations diff --git a/tests/func/remote/__init__.py b/tests/func/remote/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/func/remote/test_local.py b/tests/func/remote/test_local.py new file mode 100644 index 000000000..30d42f437 --- /dev/null +++ b/tests/func/remote/test_local.py @@ -0,0 +1,84 @@ +import os + +import mock +import pytest + +import dvc +from dvc.exceptions import DvcException +from dvc.remote.local import RemoteLOCAL +from tests.utils import trees_equal + + +class TestUnpackedDirStatusOptimization(object): + @pytest.fixture(autouse=True) + def setUp(self, dvc_repo, repo_dir): + stages = dvc_repo.add(repo_dir.DATA_DIR) + assert len(stages) == 1 + self.dir_out = stages[0].outs[0] + self.unpacked_dir = ( + self.dir_out.cache_path + RemoteLOCAL.UNPACKED_DIR_SUFFIX + ) + + def test_should_not_fail_add_on_unpacked_dir_creation_exception( + self, dvc_repo + ): + with mock.patch.object( + dvc.remote.local.RemoteLOCAL, + "_create_unpacked_dir", + side_effect=self._raise_dvc_exception, + ) as unpacked_create_spy, dvc_repo.state: + assert not dvc_repo.cache.local.changed_cache( + self.dir_out.checksum + ) + assert unpacked_create_spy.call_count == 1 + + def test_should_remove_unpacked_dir_on_create_exception(self, dvc_repo): + # artificial unpacked dir for test purpose + os.makedirs(self.unpacked_dir) + assert os.path.exists(self.unpacked_dir) + + with mock.patch.object( + dvc.remote.local.RemoteLOCAL, + "_create_unpacked_dir", + side_effect=self._raise_dvc_exception, + ), dvc_repo.state: + assert not dvc_repo.cache.local.changed_cache( + self.dir_out.checksum + ) + + assert not os.path.exists(self.unpacked_dir) + + def test_should_create_unpacked_dir_on_status_check( + self, dvc_repo, repo_dir + ): + assert not os.path.exists(self.unpacked_dir) + + with dvc_repo.state: + assert not dvc_repo.cache.local.changed_cache( + self.dir_out.checksum + ) + + assert os.path.exists(self.unpacked_dir) + + trees_equal(repo_dir.DATA_DIR, self.unpacked_dir) + + def test_should_proceed_with_full_check_on_modified_cache(self, dvc_repo): + + some_file_md5 = self.dir_out.dir_cache[0]["md5"] + assert not os.path.exists(self.unpacked_dir) + + with dvc_repo.state: + assert not dvc_repo.cache.local.changed_cache( + self.dir_out.checksum + ) + assert os.path.exists(self.unpacked_dir) + + with open(dvc_repo.cache.local.get(some_file_md5), "a") as fobj: + fobj.write("string") + + with dvc_repo.state: + assert dvc_repo.cache.local.changed_cache(self.dir_out.checksum) + + @staticmethod + def _raise_dvc_exception(*args): + raise DvcException("msg") diff --git a/tests/func/test_api.py b/tests/func/test_api.py index a5a7a5177..956a5295e 100644 --- a/tests/func/test_api.py +++ b/tests/func/test_api.py @@ -3,6 +3,7 @@ import shutil from dvc import api from dvc.main import main +from dvc.path_info import URLInfo from .test_data_cloud import ( _should_test_aws, get_aws_url, @@ -75,10 +76,8 @@ def test_get_url(repo_dir, dvc_repo, remote): run_dvc("remote", "add", "-d", "upstream", remote_url) dvc_repo.add(repo_dir.FOO) - assert api.get_url(repo_dir.FOO) == "%s/%s" % ( - remote_url, - "ac/bd18db4cc2f85cedef654fccc4a4d8", - ) + expected_url = URLInfo(remote_url) / "ac/bd18db4cc2f85cedef654fccc4a4d8" + assert api.get_url(repo_dir.FOO) == expected_url def test_open(repo_dir, dvc_repo, remote): diff --git a/tests/func/test_cache.py b/tests/func/test_cache.py index 9e0bd1dad..96297e6f3 100644 --- a/tests/func/test_cache.py +++ b/tests/func/test_cache.py @@ -71,13 +71,14 @@ class TestExternalCacheDir(TestDvc): self.assertNotEqual(len(os.listdir(cache_dir)), 0) def test_remote_references(self): - assert main(["remote", "add", "storage", "ssh://localhost"]) == 0 + ssh_url = "ssh://user@localhost:23" + assert main(["remote", "add", "storage", ssh_url]) == 0 assert main(["remote", "add", "cache", "remote://storage/tmp"]) == 0 assert main(["config", "cache.ssh", "cache"]) == 0 self.dvc.__init__() - assert self.dvc.cache.ssh.url == "ssh://localhost/tmp" + assert self.dvc.cache.ssh.path_info == ssh_url + "/tmp" class TestSharedCacheDir(TestDir): diff --git a/tests/func/test_checkout.py b/tests/func/test_checkout.py index b9ed9d264..2ac817a9f 100644 --- a/tests/func/test_checkout.py +++ b/tests/func/test_checkout.py @@ -422,7 +422,11 @@ class TestCheckoutShouldHaveSelfClearingProgressBar(TestDvc): self.write_args = [w_c[1][0] for w_c in write_calls] pattern = re.compile(".*\\[.{30}\\].*%.*") - progress_bars = [arg for arg in self.write_args if pattern.match(arg)] + progress_bars = [ + arg + for arg in self.write_args + if pattern.match(arg) and "unpacked" not in arg + ] update_bars = progress_bars[:-1] finish_bar = progress_bars[-1] diff --git a/tests/func/test_data_cloud.py b/tests/func/test_data_cloud.py index 97daa1a91..bc7a3864e 100644 --- a/tests/func/test_data_cloud.py +++ b/tests/func/test_data_cloud.py @@ -211,15 +211,6 @@ def get_gcp_url(): return "gs://" + get_gcp_storagepath() -def get_azure_url_compat(): - container_name = os.getenv("AZURE_STORAGE_CONTAINER_NAME") - connection_string = os.getenv("AZURE_STORAGE_CONNECTION_STRING") - assert container_name is not None - return "azure://ContainerName={};{}".format( - container_name, connection_string - ) - - def get_azure_url(): container_name = os.getenv("AZURE_STORAGE_CONTAINER_NAME") assert container_name is not None @@ -409,22 +400,17 @@ class TestRemoteGS(TestDataCloudBase): return RemoteGS -class TestRemoteAZURECompat(TestDataCloudBase): +class TestRemoteAZURE(TestDataCloudBase): def _should_test(self): return _should_test_azure() def _get_url(self): - return get_azure_url_compat() + return get_azure_url() def _get_cloud_class(self): return RemoteAZURE -class TestRemoteAZURE(TestRemoteAZURECompat): - def _get_url(self): - return get_azure_url() - - class TestRemoteOSS(TestDataCloudBase): def _should_test(self): return _should_test_oss() diff --git a/tests/func/test_dependency.py b/tests/func/test_dependency.py index ef046e672..67cd21539 100644 --- a/tests/func/test_dependency.py +++ b/tests/func/test_dependency.py @@ -1,3 +1,4 @@ +from __future__ import unicode_literals import copy import pytest diff --git a/tests/func/test_diff.py b/tests/func/test_diff.py index ee2fd5f5e..d7bf9ac3b 100644 --- a/tests/func/test_diff.py +++ b/tests/func/test_diff.py @@ -17,7 +17,7 @@ from tests.basic_env import TestDvc def _get_checksum(repo, file_name): outs = [out for s in repo.stages() for out in s.outs] for out in outs: - if out.url == file_name: + if out.def_path == file_name: return out.checksum diff --git a/tests/func/test_remote.py b/tests/func/test_remote.py index 9f1b07d59..0a6f74dc0 100644 --- a/tests/func/test_remote.py +++ b/tests/func/test_remote.py @@ -1,11 +1,11 @@ import os import configobj -from dvc.path.local import PathLOCAL -from dvc.remote.base import RemoteBASE from mock import patch from dvc.main import main from dvc.config import Config +from dvc.remote.base import RemoteBASE +from dvc.path_info import PathInfo from tests.basic_env import TestDvc from tests.func.test_data_cloud import get_local_storagepath @@ -182,7 +182,7 @@ def test_dir_checksum_should_be_key_order_agnostic(dvc_repo): with open(file2, "w") as fobj: fobj.write("2") - path_info = PathLOCAL(path=data_dir) + path_info = PathInfo(data_dir) with dvc_repo.state: with patch.object( RemoteBASE, diff --git a/tests/func/test_repo.py b/tests/func/test_repo.py index 43aced7c4..89880c492 100644 --- a/tests/func/test_repo.py +++ b/tests/func/test_repo.py @@ -33,9 +33,7 @@ class TestCollect(TestDvc): else: self.dvc.tree = WorkingTree() result = self.dvc.collect(target + ".dvc", with_deps=with_deps) - self.assertEqual( - [[j.rel_path for j in i.outs] for i in result], expected - ) + self.assertEqual([[str(j) for j in i.outs] for i in result], expected) return result def test(self): @@ -45,7 +43,7 @@ class TestCollect(TestDvc): "new_branch", "buzz", True, [[self.FOO], [self.BAR], ["buzz"]] ) result = self._check("new_branch", "buzz", False, [["buzz"]]) - self.assertEqual([i.rel_path for i in result[0].deps], ["bar"]) + self.assertEqual([str(i) for i in result[0].deps], ["bar"]) class TestIgnore(TestDvc): diff --git a/tests/func/test_repro.py b/tests/func/test_repro.py index a58fcbd6f..b253f3ba4 100644 --- a/tests/func/test_repro.py +++ b/tests/func/test_repro.py @@ -1,5 +1,5 @@ from __future__ import unicode_literals -from dvc.utils.compat import str, urljoin, Path +from dvc.utils.compat import str, urljoin, pathlib import os import re @@ -1062,8 +1062,10 @@ class TestReproExternalSSH(TestReproExternalBase): return "{}@127.0.0.1:{}".format(getpass.getuser(), self._dir) def cmd(self, i, o): - i = i.strip("ssh://") - o = o.strip("ssh://") + prefix = "ssh://" + assert i.startswith(prefix) and o.startswith(prefix) + i = i[len(prefix) :] + o = o[len(prefix) :] return "scp {} {}".format(i, o) def write(self, bucket, key, body): @@ -1560,7 +1562,7 @@ def foo_copy(repo_dir, dvc_repo): def test_dvc_formatting_retained(dvc_repo, foo_copy): - root = Path(dvc_repo.root_dir) + root = pathlib.Path(dvc_repo.root_dir) stage_file = root / foo_copy["stage_fname"] # Add comments and custom formatting to stage file diff --git a/tests/func/test_run.py b/tests/func/test_run.py index 047aa30b2..79ae192db 100644 --- a/tests/func/test_run.py +++ b/tests/func/test_run.py @@ -52,7 +52,7 @@ class TestRun(TestDvc): self.assertEqual(stage.cmd, cmd) self.assertEqual(len(stage.deps), len(deps)) self.assertEqual(len(stage.outs), len(outs + outs_no_cache)) - self.assertEqual(stage.outs[0].path, outs[0]) + self.assertEqual(stage.outs[0].fspath, outs[0]) self.assertEqual(stage.outs[0].checksum, file_md5(self.FOO)[0]) self.assertTrue(stage.path, fname) diff --git a/tests/func/test_s3.py b/tests/func/test_s3.py index c6768b93e..3bcb64ba3 100644 --- a/tests/func/test_s3.py +++ b/tests/func/test_s3.py @@ -1,23 +1,13 @@ -import uuid -import posixpath -from copy import copy - import boto3 import pytest -from dvc.path.s3 import PathS3 from dvc.remote.s3 import RemoteS3 -from tests.func.test_data_cloud import TEST_AWS_REPO_BUCKET, _should_test_aws +from tests.func.test_data_cloud import _should_test_aws, get_aws_url def _get_src_dst(): - prefix = str(uuid.uuid4()) - from_info = PathS3( - bucket=TEST_AWS_REPO_BUCKET, path=posixpath.join(prefix, "from") - ) - to_info = copy(from_info) - to_info.path = posixpath.join(prefix, "to") - return from_info, to_info + base_info = RemoteS3.path_cls(get_aws_url()) + return base_info / "from", base_info / "to" def test_copy_singlepart_preserve_etag(): @@ -29,7 +19,7 @@ def test_copy_singlepart_preserve_etag(): s3 = boto3.client("s3") s3.put_object(Bucket=from_info.bucket, Key=from_info.path, Body="data") - RemoteS3._copy(s3, from_info, to_info) + RemoteS3._copy(s3, from_info, to_info, {}) def _upload_multipart(s3, Bucket, Key): @@ -70,4 +60,4 @@ def test_copy_multipart_preserve_etag(): s3 = boto3.client("s3") _upload_multipart(s3, from_info.bucket, from_info.path) - RemoteS3._copy(s3, from_info, to_info) + RemoteS3._copy(s3, from_info, to_info, {}) diff --git a/tests/func/test_state.py b/tests/func/test_state.py index 70fb0db0b..ccac2139f 100644 --- a/tests/func/test_state.py +++ b/tests/func/test_state.py @@ -1,6 +1,5 @@ import os import mock -from dvc.path.local import PathLOCAL from dvc.utils.compat import str @@ -8,6 +7,7 @@ from dvc.state import State from dvc.utils import file_md5 from dvc.main import main from dvc.utils.fs import get_inode +from dvc.path_info import PathInfo from tests.basic_env import TestDvc @@ -15,7 +15,7 @@ from tests.basic_env import TestDvc class TestState(TestDvc): def test_update(self): path = os.path.join(self.dvc.root_dir, self.FOO) - path_info = PathLOCAL(path=path) + path_info = PathInfo(path) md5 = file_md5(path)[0] state = State(self.dvc, self.dvc.config.config) @@ -78,6 +78,6 @@ class TestGetStateRecordForInode(TestDvc): get_inode_mock.side_effect = self.mock_get_inode(path, inode) with state: - state.save(PathLOCAL(path=path), md5) + state.save(PathInfo(path), md5) ret = state.get_state_record_for_inode(inode) self.assertIsNotNone(ret) diff --git a/tests/unit/remote/ssh/test_ssh.py b/tests/unit/remote/ssh/test_ssh.py index cc66b3adb..b48bf082c 100644 --- a/tests/unit/remote/ssh/test_ssh.py +++ b/tests/unit/remote/ssh/test_ssh.py @@ -22,27 +22,19 @@ class TestRemoteSSH(TestCase): config = {"url": url} remote = RemoteSSH(None, config) - - self.assertEqual(remote.user, user) - self.assertEqual(remote.host, host) - self.assertEqual(remote.port, port) - self.assertEqual(remote.prefix, path) + self.assertEqual(remote.path_info, url) # SCP-like URL ssh://[user@]host.xz:/absolute/path url = "ssh://{}@{}:{}".format(user, host, path) config = {"url": url} remote = RemoteSSH(None, config) - - self.assertEqual(remote.user, user) - self.assertEqual(remote.host, host) - self.assertEqual(remote.port, remote.DEFAULT_PORT) - self.assertEqual(remote.prefix, path) + self.assertEqual(remote.path_info, url) def test_no_path(self): config = {"url": "ssh://127.0.0.1"} remote = RemoteSSH(None, config) - self.assertEqual(remote.prefix, "/") + self.assertEqual(remote.path_info.path, "") mock_ssh_config = """ @@ -79,7 +71,7 @@ def test_ssh_host_override_from_config( mock_exists.assert_called_with(RemoteSSH.ssh_config_filename()) mock_file.assert_called_with(RemoteSSH.ssh_config_filename()) - assert remote.host == expected_host + assert remote.path_info.host == expected_host @pytest.mark.parametrize( @@ -107,7 +99,7 @@ def test_ssh_user(mock_file, mock_exists, config, expected_user): mock_exists.assert_called_with(RemoteSSH.ssh_config_filename()) mock_file.assert_called_with(RemoteSSH.ssh_config_filename()) - assert remote.user == expected_user + assert remote.path_info.user == expected_user @pytest.mark.parametrize( @@ -132,7 +124,7 @@ def test_ssh_port(mock_file, mock_exists, config, expected_port): mock_exists.assert_called_with(RemoteSSH.ssh_config_filename()) mock_file.assert_called_with(RemoteSSH.ssh_config_filename()) - assert remote.port == expected_port + assert remote.path_info.port == expected_port @pytest.mark.parametrize( diff --git a/tests/unit/remote/test_azure.py b/tests/unit/remote/test_azure.py index 4ddb57ae9..e50f59cf5 100644 --- a/tests/unit/remote/test_azure.py +++ b/tests/unit/remote/test_azure.py @@ -21,9 +21,7 @@ class TestRemoteAZURE(TestCase): ) config = {"url": url} remote = RemoteAZURE(None, config) - self.assertEqual(remote.url, url) - self.assertEqual(remote.prefix, "") - self.assertEqual(remote.bucket, self.container_name) + self.assertEqual(remote.path_info, "azure://" + self.container_name) self.assertEqual(remote.connection_string, self.connection_string) def test_init(self): @@ -31,7 +29,5 @@ class TestRemoteAZURE(TestCase): url = "azure://{}/{}".format(self.container_name, prefix) config = {"url": url, "connection_string": self.connection_string} remote = RemoteAZURE(None, config) - self.assertEqual(remote.url, url) - self.assertEqual(remote.prefix, prefix) - self.assertEqual(remote.bucket, self.container_name) + self.assertEqual(remote.path_info, url) self.assertEqual(remote.connection_string, self.connection_string) diff --git a/tests/unit/remote/test_base.py b/tests/unit/remote/test_base.py index 39ed3e5ee..ee9c9835d 100644 --- a/tests/unit/remote/test_base.py +++ b/tests/unit/remote/test_base.py @@ -1,7 +1,6 @@ import mock from unittest import TestCase -from dvc.path.base import PathBASE from dvc.remote.base import RemoteBASE, RemoteCmdError, RemoteMissingDepsError @@ -44,10 +43,10 @@ class TestCacheExists(TestCase): remote = RemoteBASE(None, config) remote.PARAM_CHECKSUM = "checksum" - remote.path_info = PathBASE(None, None) + remote.path_info = remote.path_cls(config["url"]) remote.url = "" remote.prefix = "" - path_info = PathBASE("example", None) + path_info = remote.path_info / "example" checksum_info = {remote.PARAM_CHECKSUM: "1234567890"} with mock.patch.object(remote, "_checkout") as mock_checkout: diff --git a/tests/unit/remote/test_gs.py b/tests/unit/remote/test_gs.py index f4ec8c7e2..b8232806d 100644 --- a/tests/unit/remote/test_gs.py +++ b/tests/unit/remote/test_gs.py @@ -18,9 +18,7 @@ class TestRemoteGS(TestCase): def test_init(self): remote = RemoteGS(None, self.CONFIG) - self.assertEqual(remote.url, self.URL) - self.assertEqual(remote.prefix, self.PREFIX) - self.assertEqual(remote.bucket, self.BUCKET) + self.assertEqual(remote.path_info, self.URL) self.assertEqual(remote.projectname, self.PROJECT) self.assertEqual(remote.credentialpath, self.CREDENTIALPATH) diff --git a/tests/unit/remote/test_oss.py b/tests/unit/remote/test_oss.py index e29277c96..5831eebe4 100644 --- a/tests/unit/remote/test_oss.py +++ b/tests/unit/remote/test_oss.py @@ -19,9 +19,7 @@ class TestRemoteOSS(TestCase): "oss_endpoint": self.endpoint, } remote = RemoteOSS(None, config) - self.assertEqual(remote.url, url) - self.assertEqual(remote.prefix, prefix) - self.assertEqual(remote.bucket, self.bucket_name) + self.assertEqual(remote.path_info, url) self.assertEqual(remote.endpoint, self.endpoint) self.assertEqual(remote.key_id, self.key_id) self.assertEqual(remote.key_secret, self.key_secret) diff --git a/tests/unit/test_stage.py b/tests/unit/test_stage.py index 013950b04..10d6b52f2 100644 --- a/tests/unit/test_stage.py +++ b/tests/unit/test_stage.py @@ -1,5 +1,4 @@ -import os - +from dvc.path_info import PathInfo from dvc.stage import Stage import mock @@ -56,22 +55,18 @@ class TestStageChecksum(TestCase): class TestPathConversion(TestCase): def test(self): - import posixpath - import ntpath + import os stage = Stage(None, "path") - stage.wdir = posixpath.join("..", "..") - self.assertEqual(stage.dumpd()["wdir"], "../..") - - stage.wdir = ntpath.join("..", "..") + stage.wdir = os.path.join("..", "..") self.assertEqual(stage.dumpd()["wdir"], "../..") @pytest.mark.parametrize("add", [True, False]) -def test_expand_to_path_on_add_local(add): +def test_stage_fname(add): out = mock.Mock() out.is_in_repo = False - path = "path" - fname = Stage._expand_to_path_on_add_local(add, path, out, os.path) - assert fname == path + out.path_info = PathInfo("path/to/out.txt") + fname = Stage._stage_fname([out], add) + assert fname == "out.txt.dvc" diff --git a/tests/utils/__init__.py b/tests/utils/__init__.py index 3bbe0c3c3..b43af8d06 100644 --- a/tests/utils/__init__.py +++ b/tests/utils/__init__.py @@ -1,4 +1,5 @@ import os +from filecmp import dircmp from dvc.scm import Git from mock import MagicMock @@ -29,3 +30,13 @@ def cd(newdir): yield finally: os.chdir(prevdir) + + +def trees_equal(dir_path_1, dir_path_2): + + comparison = dircmp(dir_path_1, dir_path_2) + + assert set(comparison.left_only) == set(comparison.right_only) == set() + + for d in comparison.common_dirs: + trees_equal(os.path.join(dir_path_1, d), os.path.join(dir_path_2, d))