diff --git a/dvc/dependency/hdfs.py b/dvc/dependency/hdfs.py index 4ea959309..81506ba97 100644 --- a/dvc/dependency/hdfs.py +++ b/dvc/dependency/hdfs.py @@ -21,7 +21,7 @@ class DependencyHDFS(DependencyBase): user = remote.user if remote else self.group('user') self.path_info = {'scheme': 'hdfs', 'user': user, - 'url': path} + 'path': path} def changed(self): if not self.exists: diff --git a/dvc/dependency/http.py b/dvc/dependency/http.py index f05bcbed4..a2cccf454 100644 --- a/dvc/dependency/http.py +++ b/dvc/dependency/http.py @@ -21,7 +21,7 @@ class DependencyHTTP(DependencyBase): self.path_info = { 'scheme': urlparse(path).scheme, - 'url': path, + 'path': path, } def save(self): diff --git a/dvc/dependency/local.py b/dvc/dependency/local.py index b5671cbee..a9ffb6efc 100644 --- a/dvc/dependency/local.py +++ b/dvc/dependency/local.py @@ -30,7 +30,7 @@ class DependencyLOCAL(DependencyBase): path = os.path.join(remote.prefix, urlparse(path).path.lstrip('/')) if not os.path.isabs(path): - path = self.remote.ospath(path) + path = self.remote.to_ospath(path) path = os.path.join(stage.cwd, path) self.path = os.path.abspath(os.path.normpath(path)) diff --git a/dvc/remote/azure.py b/dvc/remote/azure.py index c5eac284f..515054e9b 100644 --- a/dvc/remote/azure.py +++ b/dvc/remote/azure.py @@ -1,7 +1,6 @@ from __future__ import absolute_import import os import re -import posixpath try: from azure.storage.blob import BlockBlobService @@ -85,7 +84,7 @@ class RemoteAzure(RemoteBase): return [{ 'scheme': self.scheme, 'bucket': self.bucket, - 'path': posixpath.join(self.prefix, md5[0:2], md5[2:]) + 'path': self.checksum_to_path(md5[0:2]), } for md5 in md5s] def _list_paths(self, bucket, prefix): @@ -111,9 +110,7 @@ class RemoteAzure(RemoteBase): return [] ret = len(md5s) * [False] - paths = [posixpath.join(self.prefix, - md5[0:2], - md5[2:]) for md5 in md5s] + paths = [self.checksum_to_path(md5) for md5 in md5s] for path in self._list_keys(self.bucket, self.prefix): for i, k in enumerate(paths): if k == path: @@ -193,17 +190,13 @@ class RemoteAzure(RemoteBase): if not no_progress_bar: progress.finish_target(name) - def _path_to_etag(self, path): - relpath = posixpath.relpath(path, self.prefix) - return posixpath.dirname(relpath) + posixpath.basename(relpath) - def _all(self): # NOTE: The list might be way too big(e.g. 100M entries, md5 for each # is 32 bytes, so ~3200Mb list) and we don't really need all of it at # the same time, so it makes sense to use a generator to gradually # iterate over it, without keeping all of it in memory. return ( - self._path_to_etag(path) + self.path_to_checksum(path) for path in self._list_keys(self.bucket, self.prefix) ) @@ -218,9 +211,7 @@ class RemoteAzure(RemoteBase): if etag in used: continue path_info = {'scheme': self.scheme, - 'path': posixpath.join(self.prefix, - etag[0:2], - etag[2:]), + 'path': self.checksum_to_path(etag), 'bucket': self.bucket} self.remove(path_info) removed = True diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 1d7f82128..46dcd5591 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -1,6 +1,7 @@ import os import re import errno +import posixpath from multiprocessing import cpu_count from dvc.config import Config @@ -150,3 +151,14 @@ class RemoteBase(object): assert len(names) == len(to_infos) return names + + @property + def ospath(self): + return posixpath + + def checksum_to_path(self, 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) diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index f1a3fa7e8..0c8687d88 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -1,5 +1,4 @@ import os -import posixpath try: from google.cloud import storage @@ -74,7 +73,7 @@ class RemoteGS(RemoteBase): path_info['path']) def changed_cache(self, md5): - path = posixpath.join(self.prefix, md5[0:2], md5[2:]) + path = self.checksum_to_path(md5) cache = {'scheme': 'gs', 'bucket': self.bucket, 'path': path} if {self.PARAM_MD5: md5} != self.save_info(cache): @@ -117,7 +116,7 @@ class RemoteGS(RemoteBase): raise NotImplementedError md5 = self.get_md5(path_info['bucket'], path_info['path']) - path = posixpath.join(self.prefix, md5[0:2], md5[2:]) + path = self.checksum_to_path(md5) to_info = {'scheme': 'gs', 'bucket': self.bucket, 'path': path} self._copy(path_info, to_info) @@ -151,7 +150,7 @@ class RemoteGS(RemoteBase): msg = "Checking out '{}' with cache '{}'." logger.info(msg.format(self.to_string(path_info), md5)) - path = posixpath.join(self.prefix, md5[0:2], md5[2:]) + path = self.checksum_to_path(md5) from_info = {'scheme': 'gs', 'bucket': self.bucket, 'path': path} self._copy(from_info, path_info) @@ -172,8 +171,7 @@ class RemoteGS(RemoteBase): def md5s_to_path_infos(self, md5s): return [{'scheme': 'gs', 'bucket': self.bucket, - 'path': posixpath.join(self.prefix, - md5[0:2], md5[2:])} for md5 in md5s] + 'path': self.checksum_to_path(md5)} for md5 in md5s] def _list_paths(self, bucket, prefix): for blob in self.gs.bucket(bucket).list_blobs(prefix=prefix): @@ -193,8 +191,7 @@ class RemoteGS(RemoteBase): return [] ret = len(md5s) * [False] - paths = [posixpath.join(self.prefix, - md5[0:2], md5[2:]) for md5 in md5s] + paths = [self.checksum_to_path(md5) for md5 in md5s] for path in self._list_paths(self.bucket, self.prefix): for i, k in enumerate(paths): if k == path: @@ -290,17 +287,13 @@ class RemoteGS(RemoteBase): if not no_progress_bar: progress.finish_target(name) - def _path_to_md5(self, path): - relpath = posixpath.relpath(path, self.prefix) - return posixpath.dirname(relpath) + posixpath.basename(relpath) - def _all_md5s(self): # NOTE: The list might be way too big(e.g. 100M entries, md5 for each # is 32 bytes, so ~3200Mb list) and we don't really need all of it at # the same time, so it makes sense to use a generator to gradually # iterate over it, without keeping all of it in memory. return ( - self._path_to_md5(path) + self.path_to_checksum(path) for path in self._list_paths(self.bucket, self.prefix) ) @@ -314,8 +307,7 @@ class RemoteGS(RemoteBase): continue path_info = {'scheme': 'gs', 'bucket': self.bucket, - 'path': posixpath.join(self.prefix, - md5[0:2], md5[2:])} + 'path': self.checksum_to_path(md5)} self.remove(path_info) removed = True diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py index 3b7158b2d..3db6402b8 100644 --- a/dvc/remote/hdfs.py +++ b/dvc/remote/hdfs.py @@ -22,7 +22,7 @@ class RemoteHDFS(RemoteBase): def __init__(self, project, config): self.project = project - self.url = config.get(Config.SECTION_REMOTE_URL, '/') + self.prefix = config.get(Config.SECTION_REMOTE_URL, '/') self.user = self.group('user') if not self.user: self.user = config.get(Config.SECTION_REMOTE_USER, @@ -59,38 +59,39 @@ class RemoteHDFS(RemoteBase): def checksum(self, path_info): regex = r'.*\t.*\t(?P.*)' - stdout = self.hadoop_fs('checksum {}'.format(path_info['url']), + stdout = self.hadoop_fs('checksum {}'.format(path_info['path']), user=path_info['user']) return self._group(regex, stdout, 'checksum') def cp(self, from_info, to_info): - self.hadoop_fs('mkdir -p {}'.format(posixpath.dirname(to_info['url'])), - user=to_info['user']) - self.hadoop_fs('cp -f {} {}'.format(from_info['url'], to_info['url']), + dname = posixpath.dirname(to_info['path']) + self.hadoop_fs('mkdir -p {}'.format(dname), user=to_info['user']) + self.hadoop_fs('cp -f {} {}'.format(from_info['path'], + to_info['path']), user=to_info['user']) def rm(self, path_info): - self.hadoop_fs('rm {}'.format(path_info['url']), + self.hadoop_fs('rm {}'.format(path_info['path']), user=path_info['user']) def save_info(self, path_info): if path_info['scheme'] != 'hdfs': raise NotImplementedError - assert path_info.get('url') + assert path_info.get('path') return {self.PARAM_CHECKSUM: self.checksum(path_info)} @staticmethod def to_string(path_info): return "{}://{}".format(path_info['scheme'], - path_info['url']) + path_info['path']) def changed_cache(self, checksum): cache = {} cache['scheme'] = 'hdfs' cache['user'] = self.user - cache['url'] = posixpath.join(self.url, checksum[0:2], checksum[2:]) + cache['path'] = self.checksum_to_path(checksum) if {self.PARAM_CHECKSUM: checksum} != self.save_info(cache): if self.exists(cache): @@ -118,11 +119,11 @@ class RemoteHDFS(RemoteBase): if path_info['scheme'] != 'hdfs': raise NotImplementedError - assert path_info.get('url') + assert path_info.get('path') checksum = self.checksum(path_info) dest = path_info.copy() - dest['url'] = posixpath.join(self.url, checksum[0:2], checksum[2:]) + dest['path'] = self.checksum_to_path(checksum) self.cp(path_info, dest) @@ -132,7 +133,7 @@ class RemoteHDFS(RemoteBase): if path_info['scheme'] != 'hdfs': raise NotImplementedError - assert path_info.get('url') + assert path_info.get('path') checksum = checksum_info.get(self.PARAM_CHECKSUM, None) if not checksum: @@ -158,7 +159,7 @@ class RemoteHDFS(RemoteBase): logger.info(msg.format(self.to_string(path_info), checksum)) src = path_info.copy() - src['url'] = posixpath.join(self.url, checksum[0:2], checksum[2:]) + src['path'] = self.checksum_to_path(checksum) self.cp(src, path_info) @@ -166,24 +167,23 @@ class RemoteHDFS(RemoteBase): if path_info['scheme'] != 'hdfs': raise NotImplementedError - assert path_info.get('url') + assert path_info.get('path') - logger.debug('Removing {}'.format(path_info['url'])) + logger.debug('Removing {}'.format(path_info['path'])) self.rm(path_info) def md5s_to_path_infos(self, md5s): return [{'scheme': 'hdfs', 'user': self.user, - 'url': posixpath.join(self.url, - md5[0:2], md5[2:])} for md5 in md5s] + 'path': self.checksum_to_path(md5)} for md5 in md5s] def exists(self, path_info): assert not isinstance(path_info, list) assert path_info['scheme'] == 'hdfs' try: - self.hadoop_fs('test -e {}'.format(path_info['url'])) + self.hadoop_fs('test -e {}'.format(path_info['path'])) return True except RemoteHDFSCmdError: return False @@ -213,11 +213,11 @@ class RemoteHDFS(RemoteBase): if from_info['scheme'] != 'local': raise NotImplementedError - cmd = 'mkdir -p {}'.format(posixpath.dirname(to_info['url'])) + cmd = 'mkdir -p {}'.format(posixpath.dirname(to_info['path'])) self.hadoop_fs(cmd, user=to_info['user']) cmd = 'copyFromLocal {} {}'.format(from_info['path'], - to_info['url']) + to_info['path']) self.hadoop_fs(cmd, user=to_info['user']) def download(self, @@ -242,27 +242,24 @@ class RemoteHDFS(RemoteBase): if not os.path.exists(dname): os.makedirs(dname) - cmd = 'copyToLocal {} {}'.format(from_info['url'], to_info['path']) + cmd = 'copyToLocal {} {}'.format(from_info['path'], + to_info['path']) self.hadoop_fs(cmd, user=from_info['user']) - def _path_to_checksum(self, path): - relpath = posixpath.relpath(path, self.url) - return posixpath.dirname(relpath) + posixpath.basename(relpath) - def _all_checksums(self): try: - self.hadoop_fs('test -e {}'.format(self.url)) + self.hadoop_fs('test -e {}'.format(self.prefix)) except RemoteHDFSCmdError: return [] - stdout = self.hadoop_fs('ls -R {}'.format(self.url)) + stdout = self.hadoop_fs('ls -R {}'.format(self.prefix)) lines = stdout.split('\n') flist = [] for line in lines: if not line.startswith('-'): continue flist.append(line.split()[-1]) - return [self._path_to_checksum(path) for path in flist] + return [self.path_to_checksum(path) for path in flist] def gc(self, cinfos): used = [info[self.PARAM_CHECKSUM] for info in cinfos['hdfs']] @@ -274,8 +271,7 @@ class RemoteHDFS(RemoteBase): continue path_info = {'scheme': 'hdfs', 'user': self.user, - 'url': posixpath.join(self.url, - checksum[0:2], checksum[2:])} + 'path': self.checksum_to_path(checksum)} self.remove(path_info) removed = True diff --git a/dvc/remote/http.py b/dvc/remote/http.py index 4eb6c82a9..7b2cdc380 100644 --- a/dvc/remote/http.py +++ b/dvc/remote/http.py @@ -1,7 +1,6 @@ import os import threading import requests -import posixpath from dvc.logger import logger from dvc.progress import progress @@ -53,7 +52,7 @@ class RemoteHTTP(RemoteBase): if to_info['scheme'] != 'local': raise NotImplementedError - msg = "Downloading '{}' to '{}'".format(from_info['url'], + msg = "Downloading '{}' to '{}'".format(from_info['path'], to_info['path']) logger.debug(msg) @@ -63,7 +62,7 @@ class RemoteHTTP(RemoteBase): self._makedirs(to_info['path']) - total = self._content_length(from_info['url']) + total = self._content_length(from_info['path']) if no_progress_bar or not total: cb = None @@ -71,9 +70,9 @@ class RemoteHTTP(RemoteBase): cb = ProgressBarCallback(name, total) try: - self._download_to(from_info['url'], tmp_file, callback=cb) + self._download_to(from_info['path'], tmp_file, callback=cb) except Exception as exc: - msg = "Failed to download '{}'".format(from_info['url']) + msg = "Failed to download '{}'".format(from_info['path']) logger.warn(msg, exc) continue @@ -85,13 +84,13 @@ class RemoteHTTP(RemoteBase): def exists(self, path_info): assert not isinstance(path_info, list) assert path_info['scheme'] in ['http', 'https'] - return bool(self._request('HEAD', path_info.get('url'))) + return bool(self._request('HEAD', path_info.get('path'))) def cache_exists(self, md5s): assert isinstance(md5s, list) ret = [] for md5 in md5s: - url = posixpath.join(self.prefix, md5[0:2], md5[2:]) + url = self.checksum_to_path(md5) exists = bool(self._request('HEAD', url)) ret.append(exists) return ret @@ -100,13 +99,13 @@ class RemoteHTTP(RemoteBase): if path_info['scheme'] not in ['http', 'https']: raise NotImplementedError - return {self.PARAM_ETAG: self._etag(path_info['url'])} + return {self.PARAM_ETAG: self._etag(path_info['path'])} def md5s_to_path_infos(self, md5s): return [ { 'scheme': 'http', - 'url': posixpath.join(self.prefix, md5[0:2], md5[2:]), + 'path': self.checksum_to_path(md5), } for md5 in md5s ] diff --git a/dvc/remote/local.py b/dvc/remote/local.py index d6a01599f..089c01e5f 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -77,7 +77,7 @@ class RemoteLOCAL(RemoteBase): for cache in os.listdir(subdir): path = os.path.join(subdir, cache) - clist.append(self.path_to_md5(path)) + clist.append(self.path_to_checksum(path)) return clist @@ -87,10 +87,6 @@ class RemoteLOCAL(RemoteBase): return os.path.join(self.cache_dir, md5[0:2], md5[2:]) - def path_to_md5(self, path): - relpath = os.path.relpath(path, self.cache_dir) - return os.path.dirname(relpath) + os.path.basename(relpath) - def changed_cache_file(self, md5): cache = self.get(md5) if self.state.changed(cache, md5=md5): @@ -165,8 +161,14 @@ class RemoteLOCAL(RemoteBase): raise DvcException('No possible cache types left to try out.') + @property + def ospath(self): + if os.name == 'nt': + return ntpath + return posixpath + @classmethod - def ospath(cls, path): + def to_ospath(cls, path): if os.name == 'nt': return cls.ntpath(path) return cls.unixpath(path) @@ -242,7 +244,7 @@ class RemoteLOCAL(RemoteBase): return [] for info in d: - info['relpath'] = self.ospath(info['relpath']) + info['relpath'] = self.to_ospath(info['relpath']) return d diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 0fa9869a8..afe5ae7ee 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -1,6 +1,5 @@ import os import threading -import posixpath try: import boto3 @@ -127,7 +126,7 @@ class RemoteS3(RemoteBase): raise NotImplementedError etag = self.get_etag(path_info['bucket'], path_info['path']) - path = posixpath.join(self.prefix, etag[0:2], etag[2:]) + path = self.checksum_to_path(etag) to_info = {'scheme': 's3', 'bucket': self.bucket, 'path': path} self._copy(path_info, to_info) @@ -139,7 +138,7 @@ class RemoteS3(RemoteBase): return "s3://{}/{}".format(path_info['bucket'], path_info['path']) def changed_cache(self, etag): - path = posixpath.join(self.prefix, etag[0:2], etag[2:]) + path = self.checksum_to_path(etag) cache = {'scheme': 's3', 'bucket': self.bucket, 'path': path} if {self.PARAM_ETAG: etag} != self.save_info(cache): @@ -178,7 +177,7 @@ class RemoteS3(RemoteBase): msg = "Checking out '{}' with cache '{}'." logger.info(msg.format(self.to_string(path_info), etag)) - path = posixpath.join(self.prefix, etag[0:2], etag[2:]) + path = self.checksum_to_path(etag) from_info = {'scheme': 's3', 'bucket': self.bucket, 'path': path} self._copy(from_info, path_info) @@ -196,8 +195,7 @@ class RemoteS3(RemoteBase): def md5s_to_path_infos(self, md5s): return [{'scheme': self.scheme, 'bucket': self.bucket, - 'path': posixpath.join(self.prefix, - md5[0:2], md5[2:])} for md5 in md5s] + 'path': self.checksum_to_path(md5)} for md5 in md5s] def _list_paths(self, bucket, prefix): s3 = self.s3 @@ -237,9 +235,7 @@ class RemoteS3(RemoteBase): return [] ret = len(md5s) * [False] - paths = [posixpath.join(self.prefix, - md5[0:2], - md5[2:]) for md5 in md5s] + paths = [self.checksum_to_path(md5) for md5 in md5s] for path in self._list_paths(self.bucket, self.prefix): for i, k in enumerate(paths): if k == path: @@ -338,17 +334,13 @@ class RemoteS3(RemoteBase): if not no_progress_bar: progress.finish_target(name) - def _path_to_etag(self, path): - relpath = posixpath.relpath(path, self.prefix) - return posixpath.dirname(relpath) + posixpath.basename(relpath) - def _all(self): # NOTE: The list might be way too big(e.g. 100M entries, md5 for each # is 32 bytes, so ~3200Mb list) and we don't really need all of it at # the same time, so it makes sense to use a generator to gradually # iterate over it, without keeping all of it in memory. return ( - self._path_to_etag(path) + self.path_to_checksum(path) for path in self._list_paths(self.bucket, self.prefix) ) @@ -361,8 +353,7 @@ class RemoteS3(RemoteBase): if etag in used_etags: continue path_info = {'scheme': 's3', - 'path': posixpath.join(self.prefix, - etag[0:2], etag[2:]), + 'path': self.checksum_to_path(etag), 'bucket': self.bucket} self.remove(path_info) removed = True diff --git a/dvc/remote/ssh.py b/dvc/remote/ssh.py index 922be2205..40b680393 100644 --- a/dvc/remote/ssh.py +++ b/dvc/remote/ssh.py @@ -78,8 +78,7 @@ class RemoteSSH(RemoteBase): 'host': self.host, 'user': self.user, 'port': self.port, - 'path': posixpath.join(self.prefix, - md5[0:2], md5[2:])} for md5 in md5s] + 'path': self.checksum_to_path(md5)} for md5 in md5s] def ssh(self, host=None, user=None, port=None): msg = "Establishing ssh connection with '{}' " \ @@ -244,7 +243,7 @@ class RemoteSSH(RemoteBase): cache['host'] = self.host cache['port'] = self.port cache['user'] = self.user - cache['path'] = posixpath.join(self.prefix, md5[0:2], md5[2:]) + cache['path'] = self.checksum_to_path(md5) if {self.PARAM_MD5: md5} != self.save_info(cache): if self.exists(cache): @@ -274,7 +273,7 @@ class RemoteSSH(RemoteBase): md5 = self.md5(path_info) dest = path_info.copy() - dest['path'] = posixpath.join(self.prefix, md5[0:2], md5[2:]) + dest['path'] = self.checksum_to_path(md5) self.cp(path_info, dest) @@ -308,7 +307,7 @@ class RemoteSSH(RemoteBase): logger.info(msg.format(self.to_string(path_info), md5)) src = path_info.copy() - src['path'] = posixpath.join(self.prefix, md5[0:2], md5[2:]) + src['path'] = self.checksum_to_path(md5) self.cp(src, path_info) @@ -418,10 +417,6 @@ class RemoteSSH(RemoteBase): sftp.close() ssh.close() - def _path_to_md5(self, path): - relpath = posixpath.relpath(path, self.prefix) - return posixpath.dirname(relpath) + posixpath.basename(relpath) - def _all_md5s(self): ssh = self.ssh(host=self.host, user=self.user, @@ -431,7 +426,7 @@ class RemoteSSH(RemoteBase): flist = stdout.split() ssh.close() - return [self._path_to_md5(path) for path in flist] + return [self.path_to_checksum(path) for path in flist] def gc(self, cinfos): used = [info[self.PARAM_MD5] for info in cinfos['ssh']] @@ -445,8 +440,7 @@ class RemoteSSH(RemoteBase): 'user': self.user, 'host': self.host, 'port': self.port, - 'path': posixpath.join(self.prefix, - md5[0:2], md5[2:])} + 'path': self.checksum_to_path(md5)} self.remove(path_info) removed = True diff --git a/tests/test_cache.py b/tests/test_cache.py index 7a9dd3a53..fcad3717d 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -24,8 +24,8 @@ class TestCache(TestDvc): def test_all(self): md5_list = Cache(self.dvc).local.all() self.assertEquals(len(md5_list), 2) - self.assertTrue(self.cache1_md5 in md5_list) - self.assertTrue(self.cache2_md5 in md5_list) + self.assertIn(self.cache1_md5, md5_list) + self.assertIn(self.cache2_md5, md5_list) def test_get(self): cache = Cache(self.dvc).local.get(self.cache1_md5)