Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-02 07:18:25 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'upstream/master'

Browse Source
This commit is contained in:
Sanidhya Mangal
2019-06-12 13:59:33 +05:30
parent cb4a745d1b 99d533310e
commit 771b84a9f8
64 changed files with 1113 additions and 1073 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/command/diff.py
+1 -1
View File
@@ -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
dvc/command/version.py
+3 -3
View File
@@ -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")
dvc/config.py
+2
View File
@@ -255,6 +255,7 @@ class Config(object): # pylint: disable=too-many-instance-attributes
SECTION_REMOTE_TIMEOUT = "timeout"
SECTION_REMOTE_PASSWORD = "password"
SECTION_REMOTE_ASK_PASSWORD = "ask_password"
SECTION_REMOTE_NO_TRAVERSE = "no_traverse"
SECTION_REMOTE_SCHEMA = {
SECTION_REMOTE_URL: str,
Optional(SECTION_AWS_REGION): str,
@@ -278,6 +279,7 @@ class Config(object): # pylint: disable=too-many-instance-attributes
Optional(SECTION_OSS_ACCESS_KEY_SECRET): str,
Optional(SECTION_OSS_ENDPOINT): str,
Optional(PRIVATE_CWD): str,
Optional(SECTION_REMOTE_NO_TRAVERSE, default=True): BOOL_SCHEMA,
}
SECTION_STATE = "state"
dvc/dependency/http.py
-11
View File
@@ -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)
dvc/exceptions.py
+2 -4
View File
@@ -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))
dvc/output/base.py
+40 -38
View File
@@ -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:
dvc/output/hdfs.py
-30
View File
@@ -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)
dvc/output/local.py
+38 -62
View File
@@ -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))
dvc/output/s3.py
-34
View File
@@ -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
)
dvc/output/ssh.py
-46
View File
@@ -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
)
dvc/path/__init__.py
-29
View File
@@ -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)
dvc/path/azure.py
-6
View File
@@ -1,6 +0,0 @@
from dvc.scheme import Schemes
from .base import PathCloudBASE
class PathAZURE(PathCloudBASE):
scheme = Schemes.AZURE
dvc/path/base.py
-23
View File
@@ -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
dvc/path/gs.py
-6
View File
@@ -1,6 +0,0 @@
from dvc.scheme import Schemes
from .base import PathCloudBASE
class PathGS(PathCloudBASE):
scheme = Schemes.GS
dvc/path/hdfs.py
-13
View File
@@ -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
dvc/path/http.py
-13
View File
@@ -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
dvc/path/https.py
-6
View File
@@ -1,6 +0,0 @@
from dvc.scheme import Schemes
from .http import PathHTTP
class PathHTTPS(PathHTTP):
scheme = Schemes.HTTPS
dvc/path/local.py
-17
View File
@@ -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)
dvc/path/oss.py
-6
View File
@@ -1,6 +0,0 @@
from dvc.scheme import Schemes
from .base import PathCloudBASE
class PathOSS(PathCloudBASE):
scheme = Schemes.OSS
dvc/path/s3.py
-6
View File
@@ -1,6 +0,0 @@
from dvc.scheme import Schemes
from .base import PathCloudBASE
class PathS3(PathCloudBASE):
scheme = Schemes.S3
dvc/path/ssh.py
-26
View File
@@ -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
dvc/path_info.py
+213
View File
@@ -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("/")
dvc/remote/azure.py
+37 -45
View File
@@ -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<path>[^=;]*)?|("
# backward compatibility
@@ -43,62 +44,62 @@ class RemoteAZURE(RemoteBASE):
REQUIRES = {"azure-storage-blob": BlockBlobService}
PARAM_CHECKSUM = "etag"
COPY_POLL_SECONDS = 5
SUPPORTS_NO_TRAVERSE = False
def __init__(self, repo, config):
super(RemoteAZURE, self).__init__(repo, config)
self.url = config.get(Config.SECTION_REMOTE_URL, "azure://")
match = re.match(self.REGEX, self.url) # backward compatibility
url = config.get(Config.SECTION_REMOTE_URL, "azure://")
match = re.match(self.REGEX, url) # backward compatibility
path = match.group("path")
self.bucket = (
urlparse(self.url if path else "").netloc
bucket = (
urlparse(url if path else "").netloc
or match.group("container_name") # backward compatibility
or os.getenv("AZURE_STORAGE_CONTAINER_NAME")
)
self.prefix = urlparse(self.url).path.lstrip("/") if path else ""
self.connection_string = (
config.get(Config.SECTION_AZURE_CONNECTION_STRING)
or match.group("connection_string") # backward compatibility
or os.getenv("AZURE_STORAGE_CONNECTION_STRING")
)
if not self.bucket:
if not bucket:
raise ValueError("azure storage container name missing")
if not self.connection_string:
raise ValueError("azure storage connection string missing")
self.__blob_service = None
self.path_info = PathAZURE(bucket=self.bucket)
self.path_info = (
self.path_cls(url)
if path
else self.path_cls.from_parts(scheme=self.scheme, netloc=bucket)
)
@property
def blob_service(self):
if self.__blob_service is None:
logger.debug("URL {}".format(self.url))
logger.debug("URL {}".format(self.path_info))
logger.debug("Connection string {}".format(self.connection_string))
self.__blob_service = BlockBlobService(
connection_string=self.connection_string
)
logger.debug("Container name {}".format(self.bucket))
logger.debug("Container name {}".format(self.path_info.bucket))
try: # verify that container exists
self.__blob_service.list_blobs(
self.bucket, delimiter="/", num_results=1
self.path_info.bucket, delimiter="/", num_results=1
)
except AzureMissingResourceHttpError:
self.__blob_service.create_container(self.bucket)
self.__blob_service.create_container(self.path_info.bucket)
return self.__blob_service
def remove(self, path_info):
if path_info.scheme != self.scheme:
raise NotImplementedError
logger.debug(
"Removing azure://{}/{}".format(path_info.bucket, path_info.path)
)
logger.debug("Removing {}".format(path_info))
self.blob_service.delete_blob(path_info.bucket, path_info.path)
def _list_paths(self, bucket, prefix):
@@ -118,7 +119,7 @@ class RemoteAZURE(RemoteBASE):
next_marker = blobs.next_marker
def list_cache_paths(self):
return self._list_paths(self.bucket, self.prefix)
return self._list_paths(self.path_info.bucket, self.path_info.path)
def upload(self, from_infos, to_infos, names=None, no_progress_bar=False):
names = self._verify_path_args(to_infos, from_infos, names)
@@ -130,26 +131,21 @@ class RemoteAZURE(RemoteBASE):
if from_info.scheme != "local":
raise NotImplementedError
bucket = to_info.bucket
path = to_info.path
logger.debug(
"Uploading '{}' to '{}/{}'".format(
from_info.path, bucket, path
)
)
logger.debug("Uploading '{}' to '{}'".format(from_info, to_info))
if not name:
name = os.path.basename(from_info.path)
name = from_info.name
cb = None if no_progress_bar else Callback(name)
try:
self.blob_service.create_blob_from_path(
bucket, path, from_info.path, progress_callback=cb
to_info.bucket,
to_info.path,
from_info.fspath,
progress_callback=cb,
)
except Exception:
msg = "failed to upload '{}'".format(from_info.path)
msg = "failed to upload '{}'".format(from_info)
logger.warning(msg)
else:
progress.finish_target(name)
@@ -171,32 +167,28 @@ class RemoteAZURE(RemoteBASE):
if to_info.scheme != "local":
raise NotImplementedError
bucket = from_info.bucket
path = from_info.path
logger.debug("Downloading '{}' to '{}'".format(from_info, to_info))
logger.debug(
"Downloading '{}/{}' to '{}'".format(
bucket, path, to_info.path
)
)
tmp_file = tmp_fname(to_info.path)
tmp_file = tmp_fname(to_info)
if not name:
name = os.path.basename(to_info.path)
name = to_info.name
cb = None if no_progress_bar else Callback(name)
makedirs(os.path.dirname(to_info.path), exist_ok=True)
makedirs(fspath_py35(to_info.parent), exist_ok=True)
try:
self.blob_service.get_blob_to_path(
bucket, path, tmp_file, progress_callback=cb
from_info.bucket,
from_info.path,
tmp_file,
progress_callback=cb,
)
except Exception:
msg = "failed to download '{}/{}'".format(bucket, path)
msg = "failed to download '{}'".format(from_info)
logger.warning(msg)
else:
move(tmp_file, to_info.path)
move(tmp_file, to_info.fspath)
if not no_progress_bar:
progress.finish_target(name)
dvc/remote/base.py
+115 -129
View File
@@ -1,26 +1,23 @@
from __future__ import unicode_literals
from copy import copy
from dvc.path.local import PathLOCAL
from dvc.utils.compat import str
from dvc.utils.compat import str, basestring, urlparse
import os
import re
import json
import ntpath
import logging
import tempfile
import posixpath
import itertools
from operator import itemgetter
from multiprocessing import cpu_count
from concurrent.futures import ThreadPoolExecutor
import dvc.prompt as prompt
from dvc.config import Config
from dvc.exceptions import DvcException, ConfirmRemoveError
from dvc.progress import progress
from dvc.utils import LARGE_DIR_SIZE, tmp_fname
from dvc.utils import LARGE_DIR_SIZE, tmp_fname, to_chunks
from dvc.state import StateBase
from dvc.path_info import PathInfo, URLInfo
logger = logging.getLogger(__name__)
@@ -67,9 +64,10 @@ class RemoteMissingDepsError(DvcException):
class RemoteBASE(object):
scheme = None
REGEX = None
scheme = "base"
path_cls = URLInfo
REQUIRES = {}
SUPPORTS_NO_TRAVERSE = True
JOBS = 4 * cpu_count()
PARAM_RELPATH = "relpath"
@@ -104,13 +102,14 @@ class RemoteBASE(object):
raise RemoteMissingDepsError(msg)
self.protected = False
self.no_traverse = config.get(Config.SECTION_REMOTE_NO_TRAVERSE)
self.state = StateBase()
self._dir_info = {}
def __repr__(self):
return "{class_name}: '{url}'".format(
class_name=type(self).__name__, url=(self.url or "No url")
return "{class_name}: '{path_info}'".format(
class_name=type(self).__name__,
path_info=self.path_info or "No path",
)
def compat_config(config):
@@ -118,39 +117,14 @@ class RemoteBASE(object):
@classmethod
def supported(cls, config):
url = config[Config.SECTION_REMOTE_URL]
return cls.match(url) is not None
if isinstance(config, basestring):
url = config
else:
url = config[Config.SECTION_REMOTE_URL]
@staticmethod
def _replace_sep(path, from_sep, to_sep):
assert not ntpath.isabs(path)
assert not posixpath.isabs(path)
return path.replace(from_sep, to_sep)
@classmethod
def to_posixpath(cls, path):
return cls._replace_sep(path, "\\", "/")
@classmethod
def to_ntpath(cls, path):
return cls._replace_sep(path, "/", "\\")
def to_ospath(self, path):
if self.ospath == ntpath:
return self.to_ntpath(path)
elif self.ospath == posixpath:
return self.to_posixpath(path)
assert False
@classmethod
def match(cls, url):
return re.match(cls.REGEX, url)
def group(self, name):
m = self.match(self.url)
if not m:
return None
return m.group(name)
# NOTE: silently skipping remote, calling code should handle that
parsed = urlparse(url)
return parsed.scheme == cls.scheme
@property
def cache(self):
@@ -162,28 +136,31 @@ class RemoteBASE(object):
def _collect_dir(self, path_info):
dir_info = []
p_info = copy(path_info)
dpath = p_info.path
for root, dirs, files in self.walk(path_info):
if len(files) > LARGE_DIR_SIZE:
msg = (
"Computing md5 for a large directory {}. "
"This is only done once."
)
relpath = self.ospath.relpath(root)
logger.info(msg.format(relpath))
files = progress(files, name=relpath)
title = str(self.path_cls(root))
logger.info(msg.format(title))
files = progress(files, name=title)
for fname in files:
path = self.ospath.join(root, fname)
p_info.path = path
relpath = self.to_posixpath(self.ospath.relpath(path, dpath))
checksum = self.get_file_checksum(p_info)
file_info = self.path_cls(root) / fname
relpath = file_info.relative_to(path_info)
dir_info.append(
{
self.PARAM_RELPATH: relpath,
self.PARAM_CHECKSUM: checksum,
# NOTE: this is lossy transformation:
# "hey\there" -> "hey/there"
# "hey/there" -> "hey/there"
# The latter is fine filename on Windows,
# which will transform to dir/file on back transform.
#
# Yes, this is a BUG, as long as we permit "/" in
# filenames on Windows and "\" on Unix
self.PARAM_RELPATH: relpath.as_posix(),
self.PARAM_CHECKSUM: self.get_file_checksum(file_info),
}
)
@@ -192,32 +169,27 @@ class RemoteBASE(object):
def get_dir_checksum(self, path_info):
dir_info = self._collect_dir(path_info)
checksum, from_info, to_info = self._get_dir_info_checksum(
dir_info, path_info
)
checksum, tmp_info = self._get_dir_info_checksum(dir_info)
new_info = self.cache.checksum_to_path_info(checksum)
if self.cache.changed_cache_file(checksum):
self.cache.move(from_info, to_info)
self.cache.move(tmp_info, new_info)
self.state.save(path_info, checksum)
self.state.save(to_info, checksum)
self.state.save(new_info, checksum)
return checksum
def _get_dir_info_checksum(self, dir_info, path_info):
to_info = copy(path_info)
to_info.path = self.cache.ospath.join(self.cache.prefix, tmp_fname(""))
def _get_dir_info_checksum(self, dir_info):
tmp = tempfile.NamedTemporaryFile(delete=False).name
with open(tmp, "w+") as fobj:
json.dump(dir_info, fobj, sort_keys=True)
from_info = PathLOCAL(path=tmp)
from_info = PathInfo(tmp)
to_info = self.cache.path_info / tmp_fname("")
self.cache.upload([from_info], [to_info], no_progress_bar=True)
checksum = self.get_file_checksum(to_info) + self.CHECKSUM_DIR_SUFFIX
from_info = copy(to_info)
to_info.path = self.cache.checksum_to_path(checksum)
return checksum, from_info, to_info
return checksum, to_info
def get_dir_cache(self, checksum):
assert checksum
@@ -235,7 +207,7 @@ class RemoteBASE(object):
fobj = tempfile.NamedTemporaryFile(delete=False)
path = fobj.name
to_info = PathLOCAL(path=path)
to_info = self.path_cls(path)
self.cache.download([path_info], [to_info], no_progress_bar=True)
try:
@@ -253,7 +225,9 @@ class RemoteBASE(object):
return []
for info in d:
info[self.PARAM_RELPATH] = self.to_ospath(info[self.PARAM_RELPATH])
# NOTE: here is a BUG, see comment to .as_posix() below
relpath = self.path_cls.from_posix(info[self.PARAM_RELPATH])
info[self.PARAM_RELPATH] = relpath.fspath
return d
@@ -333,7 +307,7 @@ class RemoteBASE(object):
logger.debug("'{}' hasn't changed.".format(path_info))
return False
def link(self, from_info, to_info):
def link(self, from_info, to_info, link_type=None):
self.copy(from_info, to_info)
def _save_file(self, path_info, checksum, save_link=True):
@@ -360,13 +334,9 @@ class RemoteBASE(object):
cache_info = self.checksum_to_path_info(checksum)
dir_info = self.get_dir_cache(checksum)
entry_info = copy(path_info)
for entry in dir_info:
entry_info = path_info / entry[self.PARAM_RELPATH]
entry_checksum = entry[self.PARAM_CHECKSUM]
entry_info.path = self.ospath.join(
path_info.path, entry[self.PARAM_RELPATH]
)
self._save_file(entry_info, entry_checksum, save_link=False)
self.state.save_link(path_info)
@@ -385,7 +355,8 @@ class RemoteBASE(object):
def walk(self, path_info):
raise NotImplementedError
def protect(self, path_info):
@staticmethod
def protect(path_info):
pass
def save(self, path_info, checksum_info):
@@ -450,25 +421,11 @@ class RemoteBASE(object):
return names
@property
def ospath(self):
return posixpath
def checksum_to_path(self, checksum):
assert checksum
return self.ospath.join(self.prefix, checksum[0:2], checksum[2:])
def path_to_checksum(self, path):
relpath = self.ospath.relpath(path, self.prefix)
return self.ospath.dirname(relpath) + self.ospath.basename(relpath)
return "".join(self.path_cls(path).parts[-2:])
def checksum_to_path_info(self, checksum):
path_info = copy(self.path_info)
path_info.path = self.checksum_to_path(checksum)
return path_info
def md5s_to_path_infos(self, md5s):
return [self.checksum_to_path_info(md5) for md5 in md5s]
return self.path_info / checksum[0:2] / checksum[2:]
def list_cache_paths(self):
raise NotImplementedError
@@ -483,9 +440,7 @@ class RemoteBASE(object):
)
def gc(self, cinfos):
from dvc.remote.local import RemoteLOCAL
used = {info[RemoteLOCAL.PARAM_CHECKSUM] for info in cinfos["local"]}
used = self.extract_used_local_checksums(cinfos)
if self.scheme != "":
used |= {
@@ -539,11 +494,15 @@ class RemoteBASE(object):
if self.changed_cache_file(checksum):
return True
if not self._changed_unpacked_dir(checksum):
return False
for entry in self.get_dir_cache(checksum):
checksum = entry[self.PARAM_CHECKSUM]
if self.changed_cache_file(checksum):
entry_checksum = entry[self.PARAM_CHECKSUM]
if self.changed_cache_file(entry_checksum):
return True
self._update_unpacked_dir(checksum)
return False
def changed_cache(self, checksum):
@@ -552,15 +511,36 @@ class RemoteBASE(object):
return self.changed_cache_file(checksum)
def cache_exists(self, checksums):
# NOTE: The reason for such an odd logic is that most of the remotes
# take much shorter time to just retrieve everything they have under
# a certain prefix(e.g. s3, gs, ssh, hdfs). Other remotes that can
# check if particular file exists much quicker, use their own
# implementation of cache_exists(see http, local).
#
# Result of all() might be way too big, so we should walk through
# it in one pass.
return list(filter(lambda checksum: checksum in checksums, self.all()))
"""Check if the given checksums are stored in the remote.
There are two ways of performing this check:
- Traverse: Get a list of all the files in the remote
(traversing the cache directory) and compare it with
the given checksums.
- No traverse: For each given checksum, run the `exists`
method and filter the checksums that aren't on the remote.
This is done in parallel threads.
The reason for such an odd logic is that most of the remotes
take much shorter time to just retrieve everything they have under
a certain prefix (e.g. s3, gs, ssh, hdfs). Other remotes that can
check if particular file exists much quicker, use their own
implementation of cache_exists (see http, local).
Returns:
A list with checksums that were found in the remote
"""
if self.no_traverse and self.SUPPORTS_NO_TRAVERSE:
with ThreadPoolExecutor(max_workers=self.JOBS) as executor:
path_infos = [self.checksum_to_path_info(x) for x in checksums]
chunks = to_chunks(path_infos, self.JOBS)
results = executor.map(self.exists, chunks)
in_remote = itertools.chain.from_iterable(results)
return list(itertools.compress(checksums, in_remote))
return list(set(checksums) & set(self.all()))
def already_cached(self, path_info):
current = self.get_checksum(path_info)
@@ -598,7 +578,7 @@ class RemoteBASE(object):
self.state.save_link(path_info)
self.state.save(path_info, checksum)
if progress_callback:
progress_callback.update(path_info.url)
progress_callback.update(str(path_info))
def makedirs(self, path_info):
raise NotImplementedError
@@ -615,13 +595,11 @@ class RemoteBASE(object):
logger.debug("Linking directory '{}'.".format(path_info))
entry_info = copy(path_info)
for entry in dir_info:
relpath = entry[self.PARAM_RELPATH]
entry_checksum = entry[self.PARAM_CHECKSUM]
entry_cache_info = self.checksum_to_path_info(entry_checksum)
entry_info.url = self.ospath.join(path_info.url, relpath)
entry_info.path = self.ospath.join(path_info.path, relpath)
entry_info = path_info / relpath
entry_checksum_info = {self.PARAM_CHECKSUM: entry_checksum}
if self.changed(entry_info, entry_checksum_info):
@@ -630,7 +608,7 @@ class RemoteBASE(object):
self.link(entry_cache_info, entry_info)
self.state.save(entry_info, entry_checksum)
if progress_callback:
progress_callback.update(entry_info.url)
progress_callback.update(str(entry_info))
self._remove_redundant_files(path_info, dir_info, force)
@@ -639,30 +617,22 @@ class RemoteBASE(object):
def _remove_redundant_files(self, path_info, dir_info, force):
existing_files = set(
self.ospath.join(root, fname)
self.path_cls(root) / fname
for root, _, files in self.walk(path_info)
for fname in files
)
needed_files = set(
self.ospath.join(path_info.path, entry[self.PARAM_RELPATH])
for entry in dir_info
)
needed_files = {
path_info / entry[self.PARAM_RELPATH] for entry in dir_info
}
delta = existing_files - needed_files
d_info = copy(path_info)
for path in delta:
d_info.path = path
self.safe_remove(d_info, force)
for path in existing_files - needed_files:
self.safe_remove(path, force)
def checkout(
self, path_info, checksum_info, force=False, progress_callback=None
):
if (
path_info.scheme not in ["local"]
and path_info.scheme != self.scheme
):
if path_info.scheme not in ["local", self.scheme]:
raise NotImplementedError
checksum = checksum_info.get(self.PARAM_CHECKSUM)
@@ -701,3 +671,19 @@ class RemoteBASE(object):
@staticmethod
def unprotect(path_info):
pass
def _get_unpacked_dir_names(self, checksums):
return set()
def extract_used_local_checksums(self, cinfos):
from dvc.remote import RemoteLOCAL
used = {info[RemoteLOCAL.PARAM_CHECKSUM] for info in cinfos["local"]}
unpacked = self._get_unpacked_dir_names(used)
return used | unpacked
def _changed_unpacked_dir(self, checksum):
return True
def _update_unpacked_dir(self, checksum, progress_callback=None):
pass
dvc/remote/gs.py
+49 -50
View File
@@ -1,10 +1,9 @@
from __future__ import unicode_literals
import os
import logging
import itertools
from dvc.scheme import Schemes
from dvc.path.gs import PathGS
try:
from google.cloud import storage
@@ -12,11 +11,12 @@ except ImportError:
storage = None
from dvc.utils import tmp_fname, move
from dvc.utils.compat import urlparse, makedirs
from dvc.utils.compat import makedirs, fspath_py35
from dvc.remote.base import RemoteBASE
from dvc.config import Config
from dvc.progress import progress
from dvc.exceptions import DvcException
from dvc.path_info import CloudURLInfo
logger = logging.getLogger(__name__)
@@ -24,29 +24,24 @@ logger = logging.getLogger(__name__)
class RemoteGS(RemoteBASE):
scheme = Schemes.GS
REGEX = r"^gs://(?P<path>.*)$"
path_cls = CloudURLInfo
REQUIRES = {"google.cloud.storage": storage}
PARAM_CHECKSUM = "md5"
def __init__(self, repo, config):
super(RemoteGS, self).__init__(repo, config)
storagepath = "gs://"
storagepath += config.get(Config.SECTION_AWS_STORAGEPATH, "/")
storagepath.lstrip("/")
self.url = config.get(Config.SECTION_REMOTE_URL, storagepath)
storagepath = "gs://" + config.get(Config.SECTION_GCP_STORAGEPATH, "/")
url = config.get(Config.SECTION_REMOTE_URL, storagepath)
self.path_info = self.path_cls(url)
self.projectname = config.get(Config.SECTION_GCP_PROJECTNAME, None)
self.credentialpath = config.get(Config.SECTION_GCP_CREDENTIALPATH)
parsed = urlparse(self.url)
self.bucket = parsed.netloc
self.prefix = parsed.path.lstrip("/")
self.path_info = PathGS(bucket=self.bucket)
@staticmethod
def compat_config(config):
ret = config.copy()
url = "gs://" + ret.pop(Config.SECTION_AWS_STORAGEPATH, "").lstrip("/")
url = "gs://" + ret.pop(Config.SECTION_GCP_STORAGEPATH, "").lstrip("/")
ret[Config.SECTION_REMOTE_URL] = url
return ret
@@ -89,29 +84,44 @@ class RemoteGS(RemoteBASE):
if path_info.scheme != "gs":
raise NotImplementedError
logger.debug(
"Removing gs://{}/{}".format(path_info.bucket, path_info.path)
)
logger.debug("Removing gs://{}".format(path_info))
blob = self.gs.bucket(path_info.bucket).get_blob(path_info.path)
if not blob:
return
blob.delete()
def _list_paths(self, bucket, prefix):
for blob in self.gs.bucket(bucket).list_blobs(prefix=prefix):
def _list_paths(self, bucket, prefix, gs=None):
gs = gs or self.gs
for blob in gs.bucket(bucket).list_blobs(prefix=prefix):
yield blob.name
def list_cache_paths(self):
return self._list_paths(self.bucket, self.prefix)
return self._list_paths(self.path_info.bucket, self.path_info.path)
def exists(self, path_info):
assert not isinstance(path_info, list)
assert path_info.scheme == "gs"
def exists(self, path_infos):
single_path = False
paths = self._list_paths(path_info.bucket, path_info.path)
return any(path_info.path == path for path in paths)
if not isinstance(path_infos, list):
single_path = True
path_infos = [path_infos]
gs = self.gs
paths = itertools.chain.from_iterable(
self._list_paths(path_info.bucket, path_info.path, gs)
for path_info in path_infos
)
paths = set(paths)
results = [path_info.path in paths for path_info in path_infos]
if single_path and results:
return all(results)
return results
def upload(self, from_infos, to_infos, names=None, no_progress_bar=False):
names = self._verify_path_args(to_infos, from_infos, names)
@@ -125,14 +135,10 @@ class RemoteGS(RemoteBASE):
if from_info.scheme != "local":
raise NotImplementedError
logger.debug(
"Uploading '{}' to '{}/{}'".format(
from_info.path, to_info.bucket, to_info.path
)
)
logger.debug("Uploading '{}' to '{}'".format(from_info, to_info))
if not name:
name = os.path.basename(from_info.path)
name = from_info.name
if not no_progress_bar:
progress.update_target(name, 0, None)
@@ -140,12 +146,10 @@ class RemoteGS(RemoteBASE):
try:
bucket = gs.bucket(to_info.bucket)
blob = bucket.blob(to_info.path)
blob.upload_from_filename(from_info.path)
blob.upload_from_filename(from_info.fspath)
except Exception:
msg = "failed to upload '{}' to '{}/{}'"
logger.exception(
msg.format(from_info.path, to_info.bucket, to_info.path)
)
msg = "failed to upload '{}' to '{}'"
logger.exception(msg.format(from_info, to_info))
continue
progress.finish_target(name)
@@ -173,34 +177,29 @@ class RemoteGS(RemoteBASE):
if to_info.scheme != "local":
raise NotImplementedError
msg = "Downloading '{}/{}' to '{}'".format(
from_info.bucket, from_info.path, to_info.path
)
msg = "Downloading '{}' to '{}'".format(from_info, to_info)
logger.debug(msg)
tmp_file = tmp_fname(to_info.path)
tmp_file = tmp_fname(to_info)
if not name:
name = os.path.basename(to_info.path)
name = to_info.name
if not no_progress_bar:
# percent_cb is not available for download_to_filename, so
# lets at least update progress at pathpoints(start, finish)
progress.update_target(name, 0, None)
makedirs(os.path.dirname(to_info.path), exist_ok=True)
makedirs(fspath_py35(to_info.parent), exist_ok=True)
try:
bucket = gs.bucket(from_info.bucket)
blob = bucket.get_blob(from_info.path)
blob.download_to_filename(tmp_file)
except Exception:
msg = "failed to download '{}/{}' to '{}'"
logger.exception(
msg.format(from_info.bucket, from_info.path, to_info.path)
)
msg = "failed to download '{}' to '{}'"
logger.exception(msg.format(from_info, to_info))
continue
move(tmp_file, to_info.path)
move(tmp_file, fspath_py35(to_info))
if not no_progress_bar:
progress.finish_target(name)
dvc/remote/hdfs.py
+16 -17
View File
@@ -9,7 +9,7 @@ from subprocess import Popen, PIPE
from dvc.config import Config
from dvc.scheme import Schemes
from dvc.path.hdfs import PathHDFS
from dvc.remote.base import RemoteBASE, RemoteCmdError
from dvc.utils import fix_env, tmp_fname
@@ -20,20 +20,20 @@ logger = logging.getLogger(__name__)
class RemoteHDFS(RemoteBASE):
scheme = Schemes.HDFS
REGEX = r"^hdfs://((?P<user>.*)@)?.*$"
SUPPORTS_NO_TRAVERSE = False
PARAM_CHECKSUM = "checksum"
def __init__(self, repo, config):
super(RemoteHDFS, self).__init__(repo, config)
self.url = config.get(Config.SECTION_REMOTE_URL, "/")
self.prefix = self.url
self.user = self.group("user")
url = config.get(Config.SECTION_REMOTE_URL, "/")
self.path_info = self.path_cls(url)
self.user = self.path_info.user
if not self.user:
self.user = config.get(
Config.SECTION_REMOTE_USER, getpass.getuser()
)
self.path_info = PathHDFS(user=self.user)
def hadoop_fs(self, cmd, user=None):
cmd = "hadoop fs -" + cmd
if user:
@@ -114,19 +114,18 @@ class RemoteHDFS(RemoteBASE):
raise NotImplementedError
self.hadoop_fs(
"mkdir -p {}".format(posixpath.dirname(to_info.path)),
user=to_info.user,
"mkdir -p {}".format(to_info.parent.url), user=to_info.user
)
tmp_file = tmp_fname(to_info.path)
tmp_file = tmp_fname(to_info.url)
self.hadoop_fs(
"copyFromLocal {} {}".format(from_info.path, tmp_file),
"copyFromLocal {} {}".format(from_info.fspath, tmp_file),
user=to_info.user,
)
self.hadoop_fs(
"mv {} {}".format(tmp_file, to_info.path), user=to_info.user
"mv {} {}".format(tmp_file, to_info.url), user=to_info.user
)
def download(
@@ -150,26 +149,26 @@ class RemoteHDFS(RemoteBASE):
if to_info.scheme != "local":
raise NotImplementedError
dname = os.path.dirname(to_info.path)
dname = to_info.parent.fspath
if not os.path.exists(dname):
os.makedirs(dname)
tmp_file = tmp_fname(to_info.path)
tmp_file = tmp_fname(to_info.fspath)
self.hadoop_fs(
"copyToLocal {} {}".format(from_info.path, tmp_file),
"copyToLocal {} {}".format(from_info.url, tmp_file),
user=from_info.user,
)
os.rename(tmp_file, to_info.path)
os.rename(tmp_file, to_info.fspath)
def list_cache_paths(self):
try:
self.hadoop_fs("test -e {}".format(self.prefix))
self.hadoop_fs("test -e {}".format(self.path_info.url))
except RemoteCmdError:
return []
stdout = self.hadoop_fs("ls -R {}".format(self.prefix))
stdout = self.hadoop_fs("ls -R {}".format(self.path_info.url))
lines = stdout.split("\n")
flist = []
for line in lines:
dvc/remote/http.py
+29 -24
View File
@@ -1,8 +1,8 @@
from __future__ import unicode_literals
from dvc.scheme import Schemes
from dvc.path import Path
from dvc.utils.compat import open, makedirs
from dvc.utils.compat import open, makedirs, fspath_py35
import os
import threading
@@ -34,21 +34,15 @@ class ProgressBarCallback(object):
class RemoteHTTP(RemoteBASE):
scheme = Schemes.HTTP
REGEX = r"^http://.*$"
REQUEST_TIMEOUT = 10
CHUNK_SIZE = 2 ** 16
PARAM_CHECKSUM = "etag"
def __init__(self, repo, config):
super(RemoteHTTP, self).__init__(repo, config)
self.cache_dir = config.get(Config.SECTION_REMOTE_URL)
self.url = self.cache_dir
self.path_info = Path(self.scheme)
@property
def prefix(self):
return self.cache_dir
url = config.get(Config.SECTION_REMOTE_URL)
self.path_info = self.path_cls(url) if url else None
def download(
self,
@@ -67,17 +61,15 @@ class RemoteHTTP(RemoteBASE):
if to_info.scheme != "local":
raise NotImplementedError
msg = "Downloading '{}' to '{}'".format(
from_info.path, to_info.path
)
msg = "Downloading '{}' to '{}'".format(from_info, to_info)
logger.debug(msg)
if not name:
name = os.path.basename(to_info.path)
name = to_info.name
makedirs(os.path.dirname(to_info.path), exist_ok=True)
makedirs(fspath_py35(to_info.parent), exist_ok=True)
total = self._content_length(from_info.path)
total = self._content_length(from_info.url)
if no_progress_bar or not total:
cb = None
@@ -86,27 +78,40 @@ class RemoteHTTP(RemoteBASE):
try:
self._download_to(
from_info.path, to_info.path, callback=cb, resume=resume
from_info.url, to_info.fspath, callback=cb, resume=resume
)
except Exception:
msg = "failed to download '{}'".format(from_info.path)
msg = "failed to download '{}'".format(from_info)
logger.exception(msg)
continue
if not no_progress_bar:
progress.finish_target(name)
def exists(self, path_info):
assert not isinstance(path_info, list)
assert path_info.scheme == self.scheme
return bool(self._request("HEAD", path_info.path))
def exists(self, path_infos):
single_path = False
if not isinstance(path_infos, list):
single_path = True
path_infos = [path_infos]
results = [
bool(self._request("HEAD", path_info.url))
for path_info in path_infos
]
if single_path and results:
return all(results)
return results
def cache_exists(self, md5s):
assert isinstance(md5s, list)
def func(md5):
return bool(self._request("HEAD", self.checksum_to_path(md5)))
url = self.checksum_to_path_info(md5).url
return bool(self._request("HEAD", url))
return list(filter(func, md5s))
@@ -114,7 +119,7 @@ class RemoteHTTP(RemoteBASE):
return self._request("HEAD", url).headers.get("Content-Length")
def get_file_checksum(self, path_info):
url = path_info.path
url = path_info.url
etag = self._request("HEAD", url).headers.get("ETag") or self._request(
"HEAD", url
).headers.get("Content-MD5")
dvc/remote/local/__init__.py
+173 -101
View File
@@ -3,17 +3,13 @@ from __future__ import unicode_literals
from copy import copy
from dvc.scheme import Schemes
from dvc.path.base import PathBASE
from dvc.path.local import PathLOCAL
from dvc.remote.local.slow_link_detection import slow_link_guard
from dvc.utils.compat import str, makedirs
from dvc.utils.compat import str, makedirs, fspath_py35
import os
import stat
import uuid
import ntpath
import shutil
import posixpath
import logging
from dvc.system import System
@@ -38,16 +34,19 @@ from dvc.exceptions import DvcException
from dvc.progress import progress
from concurrent.futures import ThreadPoolExecutor
from dvc.path_info import PathInfo
logger = logging.getLogger(__name__)
class RemoteLOCAL(RemoteBASE):
scheme = Schemes.LOCAL
REGEX = r"^(?P<path>.*)$"
path_cls = PathInfo
PARAM_CHECKSUM = "md5"
PARAM_PATH = "path"
UNPACKED_DIR_SUFFIX = ".unpacked"
DEFAULT_CACHE_TYPES = ["reflink", "copy"]
CACHE_TYPE_MAP = {
"copy": shutil.copyfile,
@@ -60,12 +59,6 @@ class RemoteLOCAL(RemoteBASE):
super(RemoteLOCAL, self).__init__(repo, config)
self.state = self.repo.state if self.repo else None
self.protected = config.get(Config.SECTION_CACHE_PROTECTED, False)
storagepath = config.get(Config.SECTION_AWS_STORAGEPATH, None)
self.cache_dir = config.get(Config.SECTION_REMOTE_URL, storagepath)
if self.cache_dir is not None and not os.path.isabs(self.cache_dir):
cwd = config[Config.PRIVATE_CWD]
self.cache_dir = os.path.abspath(os.path.join(cwd, self.cache_dir))
types = config.get(Config.SECTION_CACHE_TYPE, None)
if types:
@@ -75,36 +68,46 @@ class RemoteLOCAL(RemoteBASE):
else:
self.cache_types = copy(self.DEFAULT_CACHE_TYPES)
if self.cache_dir is not None and not os.path.exists(self.cache_dir):
os.mkdir(self.cache_dir)
# A clunky way to detect cache dir
storagepath = config.get(Config.SECTION_LOCAL_STORAGEPATH, None)
cache_dir = config.get(Config.SECTION_REMOTE_URL, storagepath)
if cache_dir is not None and not os.path.isabs(cache_dir):
cwd = config[Config.PRIVATE_CWD]
cache_dir = os.path.abspath(os.path.join(cwd, cache_dir))
if cache_dir is not None and not os.path.exists(cache_dir):
os.mkdir(cache_dir)
self.path_info = PathInfo(cache_dir) if cache_dir else None
self._dir_info = {}
self.path_info = PathLOCAL()
@property
def cache_dir(self):
return self.path_info.fspath if self.path_info else None
@classmethod
def supported(cls, config):
return True
@staticmethod
def compat_config(config):
ret = config.copy()
url = ret.pop(Config.SECTION_AWS_STORAGEPATH, "")
url = ret.pop(Config.SECTION_LOCAL_STORAGEPATH, "")
ret[Config.SECTION_REMOTE_URL] = url
return ret
@property
def url(self):
return self.cache_dir
@property
def prefix(self):
return self.cache_dir
def list_cache_paths(self):
clist = []
for entry in os.listdir(self.cache_dir):
subdir = os.path.join(self.cache_dir, entry)
if not os.path.isdir(subdir):
continue
assert self.path_info is not None
for cache in os.listdir(subdir):
clist.append(os.path.join(subdir, cache))
clist = []
for entry in os.listdir(fspath_py35(self.path_info)):
subdir = self.path_info / entry
if not os.path.isdir(fspath_py35(subdir)):
continue
clist.extend(
subdir / cache for cache in os.listdir(fspath_py35(subdir))
)
return clist
@@ -112,68 +115,84 @@ class RemoteLOCAL(RemoteBASE):
if not md5:
return None
return self.checksum_to_path(md5)
return self.checksum_to_path_info(md5).url
def exists(self, path_info):
assert isinstance(path_info, PathBASE)
assert path_info.scheme == "local"
return os.path.lexists(path_info.path)
return os.path.lexists(fspath_py35(path_info))
def makedirs(self, path_info):
if not self.exists(path_info):
os.makedirs(path_info.path)
os.makedirs(fspath_py35(path_info))
@slow_link_guard
def link(self, cache_info, path_info):
cache = cache_info.path
path = path_info.path
def link(self, from_info, to_info, link_type=None):
from_path = from_info.fspath
to_path = to_info.fspath
assert os.path.isfile(cache)
assert os.path.isfile(from_path)
dname = os.path.dirname(path)
dname = os.path.dirname(to_path)
if not os.path.exists(dname):
os.makedirs(dname)
# NOTE: just create an empty file for an empty cache
if os.path.getsize(cache) == 0:
open(path, "w+").close()
if os.path.getsize(from_path) == 0:
open(to_path, "w+").close()
msg = "Created empty file: {} -> {}".format(cache, path)
msg = "Created empty file: {} -> {}".format(from_path, to_path)
logger.debug(msg)
return
i = len(self.cache_types)
if not link_type:
link_types = self.cache_types
else:
link_types = [link_type]
self._try_links(from_info, to_info, link_types)
@classmethod
def _get_link_method(cls, link_type):
try:
return cls.CACHE_TYPE_MAP[link_type]
except KeyError:
raise DvcException(
"Cache type: '{}' not supported!".format(link_type)
)
def _link(self, from_info, to_info, link_method):
if self.exists(to_info):
raise DvcException("Link '{}' already exists!".format(to_info))
else:
link_method(from_info.fspath, to_info.fspath)
if self.protected:
self.protect(to_info)
msg = "Created {}'{}': {} -> {}".format(
"protected " if self.protected else "",
self.cache_types[0],
from_info,
to_info,
)
logger.debug(msg)
@slow_link_guard
def _try_links(self, from_info, to_info, link_types):
i = len(link_types)
while i > 0:
link_method = self._get_link_method(link_types[0])
try:
self.CACHE_TYPE_MAP[self.cache_types[0]](cache, path)
if self.protected:
self.protect(path_info)
msg = "Created {}'{}': {} -> {}".format(
"protected " if self.protected else "",
self.cache_types[0],
cache,
path,
)
logger.debug(msg)
self._link(from_info, to_info, link_method)
return
except DvcException as exc:
msg = "Cache type '{}' is not supported: {}"
logger.debug(msg.format(self.cache_types[0], str(exc)))
del self.cache_types[0]
logger.debug(msg.format(link_types[0], str(exc)))
del link_types[0]
i -= 1
raise DvcException("no possible cache types left to try out.")
@property
def ospath(self):
if os.name == "nt":
return ntpath
return posixpath
def already_cached(self, path_info):
assert path_info.scheme in ["", "local"]
@@ -185,7 +204,7 @@ class RemoteLOCAL(RemoteBASE):
return not self.changed_cache(current_md5)
def is_empty(self, path_info):
path = path_info.path
path = path_info.fspath
if self.isfile(path_info) and os.path.getsize(path) == 0:
return True
@@ -196,29 +215,30 @@ class RemoteLOCAL(RemoteBASE):
return False
def isfile(self, path_info):
return os.path.isfile(path_info.path)
return os.path.isfile(fspath_py35(path_info))
def isdir(self, path_info):
return os.path.isdir(path_info.path)
return os.path.isdir(fspath_py35(path_info))
def walk(self, path_info):
return os.walk(path_info.path)
return os.walk(fspath_py35(path_info))
def get_file_checksum(self, path_info):
return file_md5(path_info.path)[0]
return file_md5(fspath_py35(path_info))[0]
def remove(self, path_info):
if path_info.scheme != "local":
raise NotImplementedError
remove(path_info.path)
if self.exists(path_info):
remove(path_info.fspath)
def move(self, from_info, to_info):
if from_info.scheme != "local" or to_info.scheme != "local":
raise NotImplementedError
inp = from_info.path
outp = to_info.path
inp = from_info.fspath
outp = to_info.fspath
# moving in two stages to make the whole operation atomic in
# case inp and outp are in different filesystems and actual
@@ -228,8 +248,11 @@ class RemoteLOCAL(RemoteBASE):
move(tmp, outp)
def cache_exists(self, md5s):
assert isinstance(md5s, list)
return list(filter(lambda md5: not self.changed_cache_file(md5), md5s))
return [
checksum
for checksum in md5s
if not self.changed_cache_file(checksum)
]
def upload(self, from_infos, to_infos, names=None, no_progress_bar=False):
names = self._verify_path_args(to_infos, from_infos, names)
@@ -241,29 +264,25 @@ class RemoteLOCAL(RemoteBASE):
if from_info.scheme != "local":
raise NotImplementedError
logger.debug(
"Uploading '{}' to '{}'".format(from_info.path, to_info.path)
)
logger.debug("Uploading '{}' to '{}'".format(from_info, to_info))
if not name:
name = os.path.basename(from_info.path)
name = from_info.name
makedirs(os.path.dirname(to_info.path), exist_ok=True)
tmp_file = tmp_fname(to_info.path)
makedirs(fspath_py35(to_info.parent), exist_ok=True)
tmp_file = tmp_fname(to_info)
try:
copyfile(
from_info.path,
fspath_py35(from_info),
tmp_file,
name=name,
no_progress_bar=no_progress_bar,
)
os.rename(tmp_file, to_info.path)
os.rename(tmp_file, fspath_py35(to_info))
except Exception:
logger.exception(
"failed to upload '{}' to '{}'".format(
from_info.path, to_info.path
)
"failed to upload '{}' to '{}'".format(from_info, to_info)
)
def download(
@@ -283,28 +302,26 @@ class RemoteLOCAL(RemoteBASE):
if to_info.scheme != "local":
raise NotImplementedError
logger.debug(
"Downloading '{}' to '{}'".format(from_info.path, to_info.path)
)
logger.debug("Downloading '{}' to '{}'".format(from_info, to_info))
if not name:
name = os.path.basename(to_info.path)
name = to_info.name
makedirs(os.path.dirname(to_info.path), exist_ok=True)
tmp_file = tmp_fname(to_info.path)
makedirs(fspath_py35(to_info.parent), exist_ok=True)
tmp_file = tmp_fname(to_info)
try:
copyfile(
from_info.path,
fspath_py35(from_info),
tmp_file,
no_progress_bar=no_progress_bar,
name=name,
)
move(tmp_file, to_info.path)
move(tmp_file, fspath_py35(to_info))
except Exception:
logger.exception(
"failed to download '{}' to '{}'".format(
from_info.path, to_info.path
from_info, to_info
)
)
@@ -340,7 +357,9 @@ class RemoteLOCAL(RemoteBASE):
show_checksums=False,
download=False,
):
logger.info("Preparing to collect status from {}".format(remote.url))
logger.info(
"Preparing to collect status from {}".format(remote.path_info)
)
title = "Collecting information"
ret = {}
@@ -351,7 +370,7 @@ class RemoteLOCAL(RemoteBASE):
progress.update_target(title, 10, 100)
ret = self._group(checksum_infos, show_checksums=show_checksums)
md5s = list(ret.keys())
md5s = list(ret)
progress.update_target(title, 30, 100)
@@ -437,7 +456,7 @@ class RemoteLOCAL(RemoteBASE):
msg.format(
"download" if download else "upload",
"from" if download else "to",
remote.url,
remote.path_info,
)
)
@@ -552,7 +571,7 @@ class RemoteLOCAL(RemoteBASE):
@staticmethod
def unprotect(path_info):
path = path_info.path
path = path_info.fspath
if not os.path.exists(path):
raise DvcException(
"can't unprotect non-existing data '{}'".format(path)
@@ -565,4 +584,57 @@ class RemoteLOCAL(RemoteBASE):
@staticmethod
def protect(path_info):
os.chmod(path_info.path, stat.S_IREAD | stat.S_IRGRP | stat.S_IROTH)
os.chmod(
fspath_py35(path_info), stat.S_IREAD | stat.S_IRGRP | stat.S_IROTH
)
def _get_unpacked_dir_path_info(self, checksum):
info = self.checksum_to_path_info(checksum)
return info.with_name(info.name + self.UNPACKED_DIR_SUFFIX)
def _path_info_changed(self, path_info):
if self.exists(path_info) and self.state.get(path_info):
return False
return True
def _update_unpacked_dir(self, checksum):
unpacked_dir_info = self._get_unpacked_dir_path_info(checksum)
if not self._path_info_changed(unpacked_dir_info):
return
self.remove(unpacked_dir_info)
try:
dir_info = self.get_dir_cache(checksum)
self._create_unpacked_dir(checksum, dir_info, unpacked_dir_info)
except DvcException:
logger.warning("Could not create '{}'".format(unpacked_dir_info))
self.remove(unpacked_dir_info)
def _create_unpacked_dir(self, checksum, dir_info, unpacked_dir_info):
self.makedirs(unpacked_dir_info)
for entry in progress(dir_info, name="Created unpacked dir"):
entry_cache_info = self.checksum_to_path_info(
entry[self.PARAM_CHECKSUM]
)
relpath = entry[self.PARAM_RELPATH]
self.link(
entry_cache_info, unpacked_dir_info / relpath, "hardlink"
)
self.state.save(unpacked_dir_info, checksum)
def _changed_unpacked_dir(self, checksum):
status_unpacked_dir_info = self._get_unpacked_dir_path_info(checksum)
return not self.state.get(status_unpacked_dir_info)
def _get_unpacked_dir_names(self, checksums):
unpacked = set()
for c in checksums:
if self.is_dir_checksum(c):
unpacked.add(c + self.UNPACKED_DIR_SUFFIX)
return unpacked
dvc/remote/oss.py
+24 -44
View File
@@ -5,7 +5,6 @@ import os
import logging
from dvc.scheme import Schemes
from dvc.path.oss import PathOSS
try:
import oss2
@@ -13,11 +12,12 @@ except ImportError:
oss2 = None
from dvc.utils import tmp_fname, move
from dvc.utils.compat import urlparse, makedirs
from dvc.utils.compat import makedirs, fspath_py35
from dvc.progress import progress
from dvc.config import Config
from dvc.remote.base import RemoteBASE
from dvc.remote.azure import Callback
from dvc.path_info import CloudURLInfo
logger = logging.getLogger(__name__)
@@ -43,18 +43,17 @@ class RemoteOSS(RemoteBASE):
"""
scheme = Schemes.OSS
REGEX = r"^oss://(?P<path>.*)?$"
path_cls = CloudURLInfo
REQUIRES = {"oss2": oss2}
PARAM_CHECKSUM = "etag"
SUPPORTS_NO_TRAVERSE = False
COPY_POLL_SECONDS = 5
def __init__(self, repo, config):
super(RemoteOSS, self).__init__(repo, config)
self.url = config.get(Config.SECTION_REMOTE_URL)
parsed = urlparse(self.url)
self.bucket = parsed.netloc
self.prefix = parsed.path.lstrip("/")
url = config.get(Config.SECTION_REMOTE_URL)
self.path_info = self.path_cls(url) if url else None
self.endpoint = config.get(Config.SECTION_OSS_ENDPOINT) or os.getenv(
"OSS_ENDPOINT"
@@ -73,17 +72,17 @@ class RemoteOSS(RemoteBASE):
)
self._bucket = None
self.path_info = PathOSS(bucket=self.bucket)
@property
def oss_service(self):
if self._bucket is None:
logger.debug("URL {}".format(self.url))
logger.debug("URL {}".format(self.path_info))
logger.debug("key id {}".format(self.key_id))
logger.debug("key secret {}".format(self.key_secret))
auth = oss2.Auth(self.key_id, self.key_secret)
logger.debug("bucket name {}".format(self.bucket))
self._bucket = oss2.Bucket(auth, self.endpoint, self.bucket)
self._bucket = oss2.Bucket(
auth, self.endpoint, self.path_info.bucket
)
try: # verify that bucket exists
self._bucket.get_bucket_info()
except oss2.exceptions.NoSuchBucket:
@@ -99,10 +98,7 @@ class RemoteOSS(RemoteBASE):
if path_info.scheme != self.scheme:
raise NotImplementedError
logger.debug(
"Removing oss://{}/{}".format(path_info.bucket, path_info.path)
)
logger.debug("Removing oss://{}".format(path_info))
self.oss_service.delete_object(path_info.path)
def _list_paths(self, prefix):
@@ -110,7 +106,7 @@ class RemoteOSS(RemoteBASE):
yield blob.key
def list_cache_paths(self):
return self._list_paths(self.prefix)
return self._list_paths(self.path_info.path)
def upload(self, from_infos, to_infos, names=None, no_progress_bar=False):
names = self._verify_path_args(to_infos, from_infos, names)
@@ -122,26 +118,19 @@ class RemoteOSS(RemoteBASE):
if from_info.scheme != "local":
raise NotImplementedError
bucket = to_info.bucket
path = to_info.path
logger.debug(
"Uploading '{}' to 'oss://{}/{}'".format(
from_info.path, bucket, path
)
)
logger.debug("Uploading '{}' to '{}'".format(from_info, to_info))
if not name:
name = os.path.basename(from_info.path)
name = from_info.name
cb = None if no_progress_bar else Callback(name)
try:
self.oss_service.put_object_from_file(
path, from_info.path, progress_callback=cb
to_info.path, from_info.fspath, progress_callback=cb
)
except Exception:
msg = "failed to upload '{}'".format(from_info.path)
msg = "failed to upload '{}'".format(from_info)
logger.warning(msg)
else:
progress.finish_target(name)
@@ -161,32 +150,23 @@ class RemoteOSS(RemoteBASE):
if to_info.scheme != "local":
raise NotImplementedError
bucket = from_info.bucket
path = from_info.path
logger.debug("Downloading '{}' to '{}'".format(from_info, to_info))
logger.debug(
"Downloading 'oss://{}/{}' to '{}'".format(
bucket, path, to_info.path
)
)
tmp_file = tmp_fname(to_info.path)
tmp_file = tmp_fname(to_info)
if not name:
name = os.path.basename(to_info.path)
name = to_info.name
cb = None if no_progress_bar else Callback(name)
makedirs(os.path.dirname(to_info.path), exist_ok=True)
makedirs(fspath_py35(to_info.parent), exist_ok=True)
try:
self.oss_service.get_object_to_file(
path, tmp_file, progress_callback=cb
from_info.path, tmp_file, progress_callback=cb
)
except Exception:
msg = "failed to download 'oss://{}/{}'".format(bucket, path)
logger.warning(msg)
logger.warning("failed to download '{}'".format(from_info))
else:
move(tmp_file, to_info.path)
move(tmp_file, fspath_py35(to_info))
finally:
if not no_progress_bar:
progress.finish_target(name)
dvc/remote/s3.py
+57 -53
View File
@@ -3,9 +3,9 @@ from __future__ import unicode_literals
import os
import threading
import logging
import itertools
from dvc.scheme import Schemes
from dvc.path.s3 import PathS3
try:
import boto3
@@ -13,11 +13,12 @@ except ImportError:
boto3 = None
from dvc.utils import tmp_fname, move
from dvc.utils.compat import urlparse, makedirs
from dvc.utils.compat import makedirs, fspath_py35
from dvc.progress import progress
from dvc.config import Config
from dvc.remote.base import RemoteBASE
from dvc.exceptions import DvcException, ETagMismatchError
from dvc.path_info import CloudURLInfo
logger = logging.getLogger(__name__)
@@ -37,7 +38,7 @@ class Callback(object):
class RemoteS3(RemoteBASE):
scheme = Schemes.S3
REGEX = r"^s3://(?P<path>.*)$"
path_cls = CloudURLInfo
REQUIRES = {"boto3": boto3}
PARAM_CHECKSUM = "etag"
@@ -48,7 +49,8 @@ class RemoteS3(RemoteBASE):
config.get(Config.SECTION_AWS_STORAGEPATH, "").lstrip("/")
)
self.url = config.get(Config.SECTION_REMOTE_URL, storagepath)
url = config.get(Config.SECTION_REMOTE_URL, storagepath)
self.path_info = self.path_cls(url)
self.region = os.environ.get("AWS_DEFAULT_REGION") or config.get(
Config.SECTION_AWS_REGION
@@ -74,12 +76,6 @@ class RemoteS3(RemoteBASE):
if shared_creds:
os.environ.setdefault("AWS_SHARED_CREDENTIALS_FILE", shared_creds)
parsed = urlparse(self.url)
self.bucket = parsed.netloc
self.prefix = parsed.path.lstrip("/")
self.path_info = PathS3(bucket=self.bucket)
@staticmethod
def compat_config(config):
ret = config.copy()
@@ -117,16 +113,19 @@ class RemoteS3(RemoteBASE):
)
return obj
def _copy_multipart(self, s3, from_info, to_info, size, n_parts):
@classmethod
def _copy_multipart(
cls, s3, from_info, to_info, size, n_parts, extra_args
):
mpu = s3.create_multipart_upload(
Bucket=to_info.bucket, Key=to_info.path, **self.extra_args
Bucket=to_info.bucket, Key=to_info.path, **extra_args
)
mpu_id = mpu["UploadId"]
parts = []
byte_position = 0
for i in range(1, n_parts + 1):
obj = self.get_head_object(
obj = cls.get_head_object(
s3, from_info.bucket, from_info.path, PartNumber=i
)
part_size = obj["ContentLength"]
@@ -158,7 +157,8 @@ class RemoteS3(RemoteBASE):
MultipartUpload={"Parts": parts},
)
def _copy(self, s3, from_info, to_info):
@classmethod
def _copy(cls, s3, from_info, to_info, extra_args):
# NOTE: object's etag depends on the way it was uploaded to s3 or the
# way it was copied within the s3. More specifically, it depends on
# the chunk size that was used to transfer it, which would affect
@@ -179,41 +179,38 @@ class RemoteS3(RemoteBASE):
# preserve etag, we need to transfer each part separately, so the
# object is transfered in the same chunks as it was originally.
obj = self.get_head_object(s3, from_info.bucket, from_info.path)
obj = cls.get_head_object(s3, from_info.bucket, from_info.path)
etag = obj["ETag"].strip('"')
size = obj["ContentLength"]
_, _, parts_suffix = etag.partition("-")
if parts_suffix:
n_parts = int(parts_suffix)
self._copy_multipart(s3, from_info, to_info, size, n_parts)
cls._copy_multipart(
s3, from_info, to_info, size, n_parts, extra_args=extra_args
)
else:
source = {"Bucket": from_info.bucket, "Key": from_info.path}
s3.copy(
source, to_info.bucket, to_info.path, ExtraArgs=self.extra_args
)
s3.copy(source, to_info.bucket, to_info.path, ExtraArgs=extra_args)
cached_etag = self.get_etag(s3, to_info.bucket, to_info.path)
cached_etag = cls.get_etag(s3, to_info.bucket, to_info.path)
if etag != cached_etag:
raise ETagMismatchError(etag, cached_etag)
def copy(self, from_info, to_info, s3=None):
s3 = s3 if s3 else self.s3
self._copy(s3, from_info, to_info)
self._copy(s3, from_info, to_info, self.extra_args)
def remove(self, path_info):
if path_info.scheme != "s3":
raise NotImplementedError
logger.debug(
"Removing s3://{}/{}".format(path_info.bucket, path_info.path)
)
logger.debug("Removing {}".format(path_info))
self.s3.delete_object(Bucket=path_info.bucket, Key=path_info.path)
def _list_paths(self, bucket, prefix):
def _list_paths(self, bucket, prefix, s3=None):
""" Read config for list object api, paginate through list objects."""
s3 = self.s3
s3 = s3 or self.s3
kwargs = {"Bucket": bucket, "Prefix": prefix}
if self.list_objects:
list_objects_api = "list_objects"
@@ -228,14 +225,30 @@ class RemoteS3(RemoteBASE):
yield item["Key"]
def list_cache_paths(self):
return self._list_paths(self.bucket, self.prefix)
return self._list_paths(self.path_info.bucket, self.path_info.path)
def exists(self, path_info):
assert not isinstance(path_info, list)
assert path_info.scheme == "s3"
def exists(self, path_infos):
single_path = False
paths = self._list_paths(path_info.bucket, path_info.path)
return any(path_info.path == path for path in paths)
if not isinstance(path_infos, list):
single_path = True
path_infos = [path_infos]
s3 = self.s3
paths = itertools.chain.from_iterable(
self._list_paths(path_info.bucket, path_info.path, s3)
for path_info in path_infos
)
paths = set(paths)
results = [path_info.path in paths for path_info in path_infos]
if single_path and results:
return all(results)
return results
def upload(self, from_infos, to_infos, names=None, no_progress_bar=False):
names = self._verify_path_args(to_infos, from_infos, names)
@@ -249,28 +262,24 @@ class RemoteS3(RemoteBASE):
if from_info.scheme != "local":
raise NotImplementedError
logger.debug(
"Uploading '{}' to '{}/{}'".format(
from_info.path, to_info.bucket, to_info.path
)
)
logger.debug("Uploading '{}' to '{}'".format(from_info, to_info))
if not name:
name = os.path.basename(from_info.path)
name = from_info.name
total = os.path.getsize(from_info.path)
total = os.path.getsize(fspath_py35(from_info))
cb = None if no_progress_bar else Callback(name, total)
try:
s3.upload_file(
from_info.path,
from_info.fspath,
to_info.bucket,
to_info.path,
Callback=cb,
ExtraArgs=self.extra_args,
)
except Exception:
msg = "failed to upload '{}'".format(from_info.path)
msg = "failed to upload '{}'".format(from_info)
logger.exception(msg)
continue
@@ -299,17 +308,14 @@ class RemoteS3(RemoteBASE):
if to_info.scheme != "local":
raise NotImplementedError
msg = "Downloading '{}/{}' to '{}'".format(
from_info.bucket, from_info.path, to_info.path
)
msg = "Downloading '{}' to '{}'".format(from_info, to_info)
logger.debug(msg)
tmp_file = tmp_fname(to_info.path)
tmp_file = tmp_fname(to_info)
if not name:
name = os.path.basename(to_info.path)
makedirs(os.path.dirname(to_info.path), exist_ok=True)
name = to_info.name
makedirs(fspath_py35(to_info.parent), exist_ok=True)
try:
if no_progress_bar:
cb = None
@@ -323,13 +329,11 @@ class RemoteS3(RemoteBASE):
from_info.bucket, from_info.path, tmp_file, Callback=cb
)
except Exception:
msg = "failed to download '{}/{}'".format(
from_info.bucket, from_info.path
)
msg = "failed to download '{}'".format(from_info)
logger.exception(msg)
continue
move(tmp_file, to_info.path)
move(tmp_file, fspath_py35(to_info))
if not no_progress_bar:
progress.finish_target(name)
dvc/remote/ssh/__init__.py
+54 -47
View File
@@ -4,8 +4,7 @@ import os
import getpass
import logging
from dvc.path import Schemes
from dvc.path.ssh import PathSSH
from dvc.scheme import Schemes
try:
import paramiko
@@ -15,7 +14,7 @@ except ImportError:
import dvc.prompt as prompt
from dvc.remote.ssh.connection import SSHConnection
from dvc.config import Config
from dvc.utils.compat import urlparse
from dvc.utils.compat import urlparse, StringIO
from dvc.remote.base import RemoteBASE
@@ -24,11 +23,6 @@ logger = logging.getLogger(__name__)
class RemoteSSH(RemoteBASE):
scheme = Schemes.SSH
# NOTE: we support both URL-like (ssh://[user@]host.xz[:port]/path) and
# SCP-like (ssh://[user@]host.xz:/absolute/path) urls.
REGEX = r"^ssh://.*$"
REQUIRES = {"paramiko": paramiko}
JOBS = 4
@@ -38,27 +32,36 @@ class RemoteSSH(RemoteBASE):
def __init__(self, repo, config):
super(RemoteSSH, self).__init__(repo, config)
self.url = config.get(Config.SECTION_REMOTE_URL, "ssh://")
parsed = urlparse(self.url)
self.host = parsed.hostname
url = config.get(Config.SECTION_REMOTE_URL)
if url:
parsed = urlparse(url)
user_ssh_config = self._load_user_ssh_config(parsed.hostname)
user_ssh_config = self._load_user_ssh_config(self.host)
host = user_ssh_config.get("hostname", parsed.hostname)
user = (
config.get(Config.SECTION_REMOTE_USER)
or parsed.username
or user_ssh_config.get("user")
or getpass.getuser()
)
port = (
config.get(Config.SECTION_REMOTE_PORT)
or parsed.port
or self._try_get_ssh_config_port(user_ssh_config)
or self.DEFAULT_PORT
)
self.path_info = self.path_cls.from_parts(
scheme=self.scheme,
host=host,
user=user,
port=port,
path=parsed.path,
)
else:
self.path_info = None
user_ssh_config = {}
self.host = user_ssh_config.get("hostname", self.host)
self.user = (
config.get(Config.SECTION_REMOTE_USER)
or parsed.username
or user_ssh_config.get("user")
or getpass.getuser()
)
self.prefix = parsed.path or "/"
self.port = (
config.get(Config.SECTION_REMOTE_PORT)
or parsed.port
or self._try_get_ssh_config_port(user_ssh_config)
or self.DEFAULT_PORT
)
self.keyfile = config.get(
Config.SECTION_REMOTE_KEY_FILE
) or self._try_get_ssh_config_keyfile(user_ssh_config)
@@ -68,10 +71,6 @@ class RemoteSSH(RemoteBASE):
Config.SECTION_REMOTE_ASK_PASSWORD, False
)
self.path_info = PathSSH(
host=self.host, user=self.user, port=self.port
)
@staticmethod
def ssh_config_filename():
return os.path.expanduser(os.path.join("~", ".ssh", "config"))
@@ -81,10 +80,12 @@ class RemoteSSH(RemoteBASE):
user_config_file = RemoteSSH.ssh_config_filename()
user_ssh_config = dict()
if hostname and os.path.exists(user_config_file):
ssh_config = paramiko.SSHConfig()
with open(user_config_file) as f:
ssh_config = paramiko.SSHConfig()
ssh_config.parse(f)
user_ssh_config = ssh_config.lookup(hostname)
# For whatever reason parsing directly from f is unreliable
f_copy = StringIO(f.read())
ssh_config.parse(f_copy)
user_ssh_config = ssh_config.lookup(hostname)
return user_ssh_config
@staticmethod
@@ -128,12 +129,20 @@ class RemoteSSH(RemoteBASE):
password=self.password,
)
def exists(self, path_info):
assert not isinstance(path_info, list)
assert path_info.scheme == self.scheme
def exists(self, path_infos):
single_path = False
with self.ssh(path_info) as ssh:
return ssh.exists(path_info.path)
if not isinstance(path_infos, list):
single_path = True
path_infos = [path_infos]
with self.ssh(path_infos[0]) as ssh:
results = [ssh.exists(path_info.path) for path_info in path_infos]
if single_path and results:
return all(results)
return results
def get_file_checksum(self, path_info):
if path_info.scheme != self.scheme:
@@ -190,24 +199,22 @@ class RemoteSSH(RemoteBASE):
raise NotImplementedError
logger.debug(
"Downloading '{host}/{path}' to '{dest}'".format(
host=from_info.host, path=from_info.path, dest=to_info.path
"Downloading '{src}' to '{dest}'".format(
src=from_info, dest=to_info
)
)
try:
ssh.download(
from_info.path,
to_info.path,
to_info.fspath,
progress_title=name,
no_progress_bar=no_progress_bar,
)
except Exception:
logger.exception(
"failed to download '{host}/{path}' to '{dest}'".format(
host=from_info.host,
path=from_info.path,
dest=to_info.path,
"failed to download '{src}' to '{dest}'".format(
src=from_info, dest=to_info
)
)
continue
@@ -227,7 +234,7 @@ class RemoteSSH(RemoteBASE):
try:
ssh.upload(
from_info.path,
from_info.fspath,
to_info.path,
progress_title=name,
no_progress_bar=no_progress_bar,
@@ -241,7 +248,7 @@ class RemoteSSH(RemoteBASE):
def list_cache_paths(self):
with self.ssh(self.path_info) as ssh:
return list(ssh.walk_files(self.prefix))
return list(ssh.walk_files(self.path_info.path))
def walk(self, path_info):
with self.ssh(path_info) as ssh:
dvc/repo/__init__.py
+17 -17
View File
@@ -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)
dvc/repo/checkout.py
+6 -4
View File
@@ -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)
dvc/repo/metrics/show.py
+7 -11
View File
@@ -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
dvc/repo/move.py
+3 -2
View File
@@ -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)
dvc/repo/pkg/install.py
+1 -1
View File
@@ -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()
dvc/scm/git/temp_git_repo.py
+4 -2
View File
@@ -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)
dvc/stage.py
+22 -40
View File
@@ -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)
dvc/state.py
+4 -3
View File
@@ -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
dvc/utils/__init__.py
+2 -1
View File
@@ -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():
dvc/utils/collections.py
-5
View File
@@ -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.
dvc/utils/compat.py
+44 -2
View File
@@ -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
setup.py
+1
View File
@@ -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
tests/func/remote/__init__.py
View File
tests/func/remote/test_local.py
+84
View File
@@ -0,0 +1,84 @@
import os
import mock
import pytest
import dvc
from dvc.exceptions import DvcException
from dvc.remote.local import RemoteLOCAL
from tests.utils import trees_equal
class TestUnpackedDirStatusOptimization(object):
@pytest.fixture(autouse=True)
def setUp(self, dvc_repo, repo_dir):
stages = dvc_repo.add(repo_dir.DATA_DIR)
assert len(stages) == 1
self.dir_out = stages[0].outs[0]
self.unpacked_dir = (
self.dir_out.cache_path + RemoteLOCAL.UNPACKED_DIR_SUFFIX
)
def test_should_not_fail_add_on_unpacked_dir_creation_exception(
self, dvc_repo
):
with mock.patch.object(
dvc.remote.local.RemoteLOCAL,
"_create_unpacked_dir",
side_effect=self._raise_dvc_exception,
) as unpacked_create_spy, dvc_repo.state:
assert not dvc_repo.cache.local.changed_cache(
self.dir_out.checksum
)
assert unpacked_create_spy.call_count == 1
def test_should_remove_unpacked_dir_on_create_exception(self, dvc_repo):
# artificial unpacked dir for test purpose
os.makedirs(self.unpacked_dir)
assert os.path.exists(self.unpacked_dir)
with mock.patch.object(
dvc.remote.local.RemoteLOCAL,
"_create_unpacked_dir",
side_effect=self._raise_dvc_exception,
), dvc_repo.state:
assert not dvc_repo.cache.local.changed_cache(
self.dir_out.checksum
)
assert not os.path.exists(self.unpacked_dir)
def test_should_create_unpacked_dir_on_status_check(
self, dvc_repo, repo_dir
):
assert not os.path.exists(self.unpacked_dir)
with dvc_repo.state:
assert not dvc_repo.cache.local.changed_cache(
self.dir_out.checksum
)
assert os.path.exists(self.unpacked_dir)
trees_equal(repo_dir.DATA_DIR, self.unpacked_dir)
def test_should_proceed_with_full_check_on_modified_cache(self, dvc_repo):
some_file_md5 = self.dir_out.dir_cache[0]["md5"]
assert not os.path.exists(self.unpacked_dir)
with dvc_repo.state:
assert not dvc_repo.cache.local.changed_cache(
self.dir_out.checksum
)
assert os.path.exists(self.unpacked_dir)
with open(dvc_repo.cache.local.get(some_file_md5), "a") as fobj:
fobj.write("string")
with dvc_repo.state:
assert dvc_repo.cache.local.changed_cache(self.dir_out.checksum)
@staticmethod
def _raise_dvc_exception(*args):
raise DvcException("msg")
tests/func/test_api.py
+3 -4
View File
@@ -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):
tests/func/test_cache.py
+3 -2
View File
@@ -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):
tests/func/test_checkout.py
+5 -1
View File
@@ -422,7 +422,11 @@ class TestCheckoutShouldHaveSelfClearingProgressBar(TestDvc):
self.write_args = [w_c[1][0] for w_c in write_calls]
pattern = re.compile(".*\\[.{30}\\].*%.*")
progress_bars = [arg for arg in self.write_args if pattern.match(arg)]
progress_bars = [
arg
for arg in self.write_args
if pattern.match(arg) and "unpacked" not in arg
]
update_bars = progress_bars[:-1]
finish_bar = progress_bars[-1]
tests/func/test_data_cloud.py
+2 -16
View File
@@ -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()
tests/func/test_dependency.py
+1
View File
@@ -1,3 +1,4 @@
from __future__ import unicode_literals
import copy
import pytest
tests/func/test_diff.py
+1 -1
View File
@@ -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
tests/func/test_remote.py
+3 -3
View File
@@ -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,
tests/func/test_repo.py
+2 -4
View File
@@ -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):
tests/func/test_repro.py
+6 -4
View File
@@ -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
tests/func/test_run.py
+1 -1
View File
@@ -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)
tests/func/test_s3.py
+5 -15
View File
@@ -1,23 +1,13 @@
import uuid
import posixpath
from copy import copy
import boto3
import pytest
from dvc.path.s3 import PathS3
from dvc.remote.s3 import RemoteS3
from tests.func.test_data_cloud import TEST_AWS_REPO_BUCKET, _should_test_aws
from tests.func.test_data_cloud import _should_test_aws, get_aws_url
def _get_src_dst():
prefix = str(uuid.uuid4())
from_info = PathS3(
bucket=TEST_AWS_REPO_BUCKET, path=posixpath.join(prefix, "from")
)
to_info = copy(from_info)
to_info.path = posixpath.join(prefix, "to")
return from_info, to_info
base_info = RemoteS3.path_cls(get_aws_url())
return base_info / "from", base_info / "to"
def test_copy_singlepart_preserve_etag():
@@ -29,7 +19,7 @@ def test_copy_singlepart_preserve_etag():
s3 = boto3.client("s3")
s3.put_object(Bucket=from_info.bucket, Key=from_info.path, Body="data")
RemoteS3._copy(s3, from_info, to_info)
RemoteS3._copy(s3, from_info, to_info, {})
def _upload_multipart(s3, Bucket, Key):
@@ -70,4 +60,4 @@ def test_copy_multipart_preserve_etag():
s3 = boto3.client("s3")
_upload_multipart(s3, from_info.bucket, from_info.path)
RemoteS3._copy(s3, from_info, to_info)
RemoteS3._copy(s3, from_info, to_info, {})
tests/func/test_state.py
+3 -3
View File
@@ -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)
tests/unit/remote/ssh/test_ssh.py
+6 -14
View File
@@ -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(
tests/unit/remote/test_azure.py
+2 -6
View File
@@ -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)
tests/unit/remote/test_base.py
+2 -3
View File
@@ -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:
tests/unit/remote/test_gs.py
+1 -3
View File
@@ -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)
tests/unit/remote/test_oss.py
+1 -3
View File
@@ -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)
tests/unit/test_stage.py
+7 -12
View File
@@ -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"
tests/utils/__init__.py
+11
View File
@@ -1,4 +1,5 @@
import os
from filecmp import dircmp
from dvc.scm import Git
from mock import MagicMock
@@ -29,3 +30,13 @@ def cd(newdir):
yield
finally:
os.chdir(prevdir)
def trees_equal(dir_path_1, dir_path_2):
comparison = dircmp(dir_path_1, dir_path_2)
assert set(comparison.left_only) == set(comparison.right_only) == set()
for d in comparison.common_dirs:
trees_equal(os.path.join(dir_path_1, d), os.path.join(dir_path_2, d))