From d80293c44d49713ed84c395a914b13d44688ee1b Mon Sep 17 00:00:00 2001 From: Alexander Schepanovski Date: Thu, 30 May 2019 23:03:00 +0700 Subject: [PATCH] dvc: path info refactoring Replaced Path* dataclasses with PathInfo/URLInfo/CloudURLInfo, which know how parse, serialize, stringify and join themselves. They also provide useful accessors like part of an url or pathlib.Path like .parent or .name. Along the way two families of bugs where fixed and should never occur in the future: - paths not matched as strings anymore, e.g. using str.startswith() - Output.path is eliminated as ambiguous Both families might had caused undefined surprising behavior when local and remote paths clash. --- dvc/command/diff.py | 2 +- dvc/command/version.py | 6 +- dvc/dependency/http.py | 11 -- dvc/exceptions.py | 6 +- dvc/output/base.py | 78 +++++------ dvc/output/hdfs.py | 30 ----- dvc/output/local.py | 100 ++++++-------- dvc/output/s3.py | 34 ----- dvc/output/ssh.py | 46 ------- dvc/path/__init__.py | 29 ---- dvc/path/azure.py | 6 - dvc/path/base.py | 23 ---- dvc/path/gs.py | 6 - dvc/path/hdfs.py | 13 -- dvc/path/http.py | 13 -- dvc/path/https.py | 6 - dvc/path/local.py | 17 --- dvc/path/oss.py | 6 - dvc/path/s3.py | 6 - dvc/path/ssh.py | 26 ---- dvc/path_info.py | 213 ++++++++++++++++++++++++++++++ dvc/remote/azure.py | 81 +++++------- dvc/remote/base.py | 165 ++++++++--------------- dvc/remote/gs.py | 66 ++++----- dvc/remote/hdfs.py | 32 +++-- dvc/remote/http.py | 35 ++--- dvc/remote/local/__init__.py | 187 ++++++++++++-------------- dvc/remote/oss.py | 67 ++++------ dvc/remote/s3.py | 53 +++----- dvc/remote/ssh/__init__.py | 83 ++++++------ dvc/repo/__init__.py | 34 ++--- dvc/repo/checkout.py | 10 +- dvc/repo/metrics/show.py | 18 +-- dvc/repo/move.py | 5 +- dvc/repo/pkg/install.py | 2 +- dvc/scm/git/temp_git_repo.py | 6 +- dvc/stage.py | 62 +++------ dvc/state.py | 7 +- dvc/utils/__init__.py | 3 +- dvc/utils/collections.py | 5 - dvc/utils/compat.py | 46 ++++++- setup.py | 1 + tests/func/remote/test_local.py | 4 +- tests/func/test_api.py | 7 +- tests/func/test_cache.py | 5 +- tests/func/test_data_cloud.py | 18 +-- tests/func/test_dependency.py | 1 + tests/func/test_diff.py | 2 +- tests/func/test_remote.py | 6 +- tests/func/test_repo.py | 6 +- tests/func/test_repro.py | 10 +- tests/func/test_run.py | 2 +- tests/func/test_s3.py | 16 +-- tests/func/test_state.py | 6 +- tests/unit/remote/ssh/test_ssh.py | 20 +-- tests/unit/remote/test_azure.py | 8 +- tests/unit/remote/test_base.py | 5 +- tests/unit/remote/test_gs.py | 4 +- tests/unit/remote/test_oss.py | 4 +- tests/unit/test_stage.py | 19 +-- 60 files changed, 762 insertions(+), 1026 deletions(-) delete mode 100644 dvc/path/__init__.py delete mode 100644 dvc/path/azure.py delete mode 100644 dvc/path/base.py delete mode 100644 dvc/path/gs.py delete mode 100644 dvc/path/hdfs.py delete mode 100644 dvc/path/http.py delete mode 100644 dvc/path/https.py delete mode 100644 dvc/path/local.py delete mode 100644 dvc/path/oss.py delete mode 100644 dvc/path/s3.py delete mode 100644 dvc/path/ssh.py create mode 100644 dvc/path_info.py 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/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..913dc62c9 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 @@ -47,58 +48,57 @@ class RemoteAZURE(RemoteBASE): 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 +118,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 +130,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 +166,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 0074351f8..dacb84e00 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -1,17 +1,11 @@ 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 from operator import itemgetter from multiprocessing import cpu_count @@ -21,6 +15,7 @@ from dvc.exceptions import DvcException, ConfirmRemoveError from dvc.progress import progress from dvc.utils import LARGE_DIR_SIZE, tmp_fname from dvc.state import StateBase +from dvc.path_info import PathInfo, URLInfo logger = logging.getLogger(__name__) @@ -67,8 +62,8 @@ class RemoteMissingDepsError(DvcException): class RemoteBASE(object): - scheme = None - REGEX = None + scheme = "base" + path_cls = URLInfo REQUIRES = {} JOBS = 4 * cpu_count() @@ -109,8 +104,9 @@ class RemoteBASE(object): 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 +114,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 +133,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 +166,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 +204,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 +222,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 @@ -360,13 +331,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) @@ -451,25 +418,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 @@ -601,7 +554,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 @@ -618,13 +571,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): @@ -633,7 +584,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) @@ -642,30 +593,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) diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index 2e747f1dd..dbc40e3b6 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -1,10 +1,8 @@ from __future__ import unicode_literals -import os import logging from dvc.scheme import Schemes -from dvc.path.gs import PathGS try: from google.cloud import storage @@ -12,11 +10,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 +23,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,10 +83,7 @@ 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 @@ -104,7 +95,7 @@ class RemoteGS(RemoteBASE): 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) @@ -125,14 +116,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 +127,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 +158,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..27d710786 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 @@ -24,16 +24,15 @@ class RemoteHDFS(RemoteBASE): 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 +113,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 +148,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..81734773c 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,11 +78,11 @@ 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 @@ -100,13 +92,14 @@ class RemoteHTTP(RemoteBASE): 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)) + return bool(self._request("HEAD", path_info.url)) 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 +107,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 3e50591d2..afe15a602 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 @@ -35,16 +31,17 @@ from dvc.utils import ( ) from dvc.config import Config from dvc.exceptions import DvcException -from dvc.progress import progress, ProgressCallback +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" @@ -62,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: @@ -77,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 @@ -114,20 +115,19 @@ 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)) def link(self, from_info, to_info, link_type=None): - from_path = from_info.path - to_path = to_info.path + from_path = from_info.fspath + to_path = to_info.fspath assert os.path.isfile(from_path) @@ -160,12 +160,10 @@ class RemoteLOCAL(RemoteBASE): ) def _link(self, from_info, to_info, link_method): - if os.path.lexists(to_info.path): - raise DvcException( - "Link '{}' already exists!".format(to_info.path) - ) + if self.exists(to_info): + raise DvcException("Link '{}' already exists!".format(to_info)) else: - link_method(from_info.path, to_info.path) + link_method(from_info.fspath, to_info.fspath) if self.protected: self.protect(to_info) @@ -173,10 +171,9 @@ class RemoteLOCAL(RemoteBASE): msg = "Created {}'{}': {} -> {}".format( "protected " if self.protected else "", self.cache_types[0], - from_info.path, - to_info.path, + from_info, + to_info, ) - logger.debug(msg) @slow_link_guard @@ -196,12 +193,6 @@ class RemoteLOCAL(RemoteBASE): 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"] @@ -213,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 @@ -224,30 +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 if self.exists(path_info): - remove(path_info.path) + 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 @@ -270,29 +261,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( @@ -312,28 +299,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 ) ) @@ -369,7 +354,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 = {} @@ -466,7 +453,7 @@ class RemoteLOCAL(RemoteBASE): msg.format( "download" if download else "upload", "from" if download else "to", - remote.url, + remote.path_info, ) ) @@ -581,7 +568,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) @@ -594,19 +581,13 @@ class RemoteLOCAL(RemoteBASE): @staticmethod def protect(path_info): - os.chmod(path_info.path, stat.S_IREAD | stat.S_IRGRP | stat.S_IROTH) - - @classmethod - def _append_unpacked_suffix(cls, string): - return string + cls.UNPACKED_DIR_SUFFIX - - def _get_unpacked_dir_path_info(self, checksum): - unpacked_dir_info = self.checksum_to_path_info(checksum) - unpacked_dir_info.path = self._append_unpacked_suffix( - unpacked_dir_info.path + os.chmod( + fspath_py35(path_info), stat.S_IREAD | stat.S_IRGRP | stat.S_IROTH ) - return unpacked_dir_info + 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): @@ -625,28 +606,22 @@ class RemoteLOCAL(RemoteBASE): 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.path) - ) + 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): - progress_callback = ProgressCallback(len(dir_info)) self.makedirs(unpacked_dir_info) - for index, entry in enumerate(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] - - unpacked_entry_info = copy(entry_cache_info) - unpacked_entry_info.path = self.ospath.join( - unpacked_dir_info.path, relpath + self.link( + entry_cache_info, unpacked_dir_info / relpath, "hardlink" ) - self.link(entry_cache_info, unpacked_entry_info, "hardlink") - progress_callback.update("Creating unpacked dir.") self.state.save(unpacked_dir_info, checksum) def _changed_unpacked_dir(self, checksum): @@ -658,5 +633,5 @@ class RemoteLOCAL(RemoteBASE): unpacked = set() for c in checksums: if self.is_dir_checksum(c): - unpacked.add(self._append_unpacked_suffix(c)) + unpacked.add(c + self.UNPACKED_DIR_SUFFIX) return unpacked diff --git a/dvc/remote/oss.py b/dvc/remote/oss.py index 56879501d..2270f9404 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,7 +43,7 @@ class RemoteOSS(RemoteBASE): """ scheme = Schemes.OSS - REGEX = r"^oss://(?P.*)?$" + path_cls = CloudURLInfo REQUIRES = {"oss2": oss2} PARAM_CHECKSUM = "etag" COPY_POLL_SECONDS = 5 @@ -51,10 +51,8 @@ class RemoteOSS(RemoteBASE): 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 +71,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 +97,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 +105,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 +117,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 +149,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..5fce4336c 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -5,7 +5,6 @@ import threading import logging from dvc.scheme import Schemes -from dvc.path.s3 import PathS3 try: import boto3 @@ -13,11 +12,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 +37,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 +48,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 +75,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() @@ -205,10 +200,7 @@ class RemoteS3(RemoteBASE): 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): @@ -228,7 +220,7 @@ 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) @@ -249,28 +241,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 +287,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 +308,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..205163adf 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 @@ -190,24 +191,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 +226,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 +240,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/test_local.py b/tests/func/remote/test_local.py index e26b2061f..30d42f437 100644 --- a/tests/func/remote/test_local.py +++ b/tests/func/remote/test_local.py @@ -15,8 +15,8 @@ class TestUnpackedDirStatusOptimization(object): stages = dvc_repo.add(repo_dir.DATA_DIR) assert len(stages) == 1 self.dir_out = stages[0].outs[0] - self.unpacked_dir = RemoteLOCAL._append_unpacked_suffix( - self.dir_out.cache_path + self.unpacked_dir = ( + self.dir_out.cache_path + RemoteLOCAL.UNPACKED_DIR_SUFFIX ) def test_should_not_fail_add_on_unpacked_dir_creation_exception( 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_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..265a8820c 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(): 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"