From 933d524ebe57cedd5a25f6b6e814e69f8b9f4fef Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Thu, 20 Dec 2018 13:32:12 +0200 Subject: [PATCH] remote: RFC: use 'path' instead of 'key' Signed-off-by: Ruslan Kuprieiev --- dvc/dependency/gs.py | 6 ++-- dvc/dependency/s3.py | 6 ++-- dvc/remote/azure.py | 44 +++++++++++++------------- dvc/remote/gs.py | 75 ++++++++++++++++++++++---------------------- dvc/remote/s3.py | 75 +++++++++++++++++++++++--------------------- 5 files changed, 106 insertions(+), 100 deletions(-) diff --git a/dvc/dependency/gs.py b/dvc/dependency/gs.py index 2c0be604e..6f31f3d9e 100644 --- a/dvc/dependency/gs.py +++ b/dvc/dependency/gs.py @@ -17,9 +17,9 @@ class DependencyGS(DependencyS3): self.info = info self.remote = remote if remote else RemoteGS(stage.project, {}) bucket = remote.bucket if remote else urlparse(path).netloc - key = urlparse(path).path.lstrip('/') + path = urlparse(path).path.lstrip('/') if remote: - key = posixpath.join(remote.prefix, key) + path = posixpath.join(remote.prefix, path) self.path_info = {'scheme': 'gs', 'bucket': bucket, - 'key': key} + 'path': path} diff --git a/dvc/dependency/s3.py b/dvc/dependency/s3.py index 9eb2a48d0..d2f0c4eef 100644 --- a/dvc/dependency/s3.py +++ b/dvc/dependency/s3.py @@ -18,12 +18,12 @@ class DependencyS3(DependencyBase): self.remote = remote or RemoteS3(stage.project, {}) bucket = remote.bucket if remote else urlparse(path).netloc - key = urlparse(path).path.lstrip('/') + path = urlparse(path).path.lstrip('/') if remote: - key = posixpath.join(remote.prefix, key) + path = posixpath.join(remote.prefix, path) self.path_info = {'scheme': 's3', 'bucket': bucket, - 'key': key} + 'path': path} def changed(self): if not self.exists: diff --git a/dvc/remote/azure.py b/dvc/remote/azure.py index f781dcfc2..c5eac284f 100644 --- a/dvc/remote/azure.py +++ b/dvc/remote/azure.py @@ -77,18 +77,18 @@ class RemoteAzure(RemoteBase): raise NotImplementedError logger.debug('Removing azure://{}/{}'.format(path_info['bucket'], - path_info['key'])) + path_info['path'])) - self.blob_service.delete_blob(path_info['bucket'], path_info['key']) + self.blob_service.delete_blob(path_info['bucket'], path_info['path']) def md5s_to_path_infos(self, md5s): return [{ 'scheme': self.scheme, 'bucket': self.bucket, - 'key': posixpath.join(self.prefix, md5[0:2], md5[2:]) + 'path': posixpath.join(self.prefix, md5[0:2], md5[2:]) } for md5 in md5s] - def _list_keys(self, bucket, prefix): + def _list_paths(self, bucket, prefix): blob_service = self.blob_service next_marker = None while True: @@ -111,13 +111,15 @@ class RemoteAzure(RemoteBase): return [] ret = len(md5s) * [False] - keys = [posixpath.join(self.prefix, md5[0:2], md5[2:]) for md5 in md5s] - for key in self._list_keys(self.bucket, self.prefix): - for i, k in enumerate(keys): - if k == key: + paths = [posixpath.join(self.prefix, + md5[0:2], + md5[2:]) for md5 in md5s] + for path in self._list_keys(self.bucket, self.prefix): + for i, k in enumerate(paths): + if k == path: ret[i] = True - assert len(ret) == len(keys) == len(md5s) + assert len(ret) == len(paths) == len(md5s) return ret @@ -132,10 +134,10 @@ class RemoteAzure(RemoteBase): raise NotImplementedError bucket = to_info['bucket'] - key = to_info['key'] + path = to_info['path'] logger.debug("Uploading '{}' to '{}/{}'".format( - from_info['path'], bucket, key)) + from_info['path'], bucket, path)) if not name: name = os.path.basename(from_info['path']) @@ -144,7 +146,7 @@ class RemoteAzure(RemoteBase): try: self.blob_service.create_blob_from_path( - bucket, key, from_info['path'], progress_callback=cb) + bucket, path, from_info['path'], progress_callback=cb) except Exception as ex: msg = "Failed to upload '{}'".format(from_info['path']) logger.warn(msg, ex) @@ -166,10 +168,10 @@ class RemoteAzure(RemoteBase): raise NotImplementedError bucket = from_info['bucket'] - key = from_info['key'] + path = from_info['path'] logger.debug("Downloading '{}/{}' to '{}'".format( - bucket, key, to_info['path'])) + bucket, path, to_info['path'])) tmp_file = self.tmp_file(to_info['path']) if not name: @@ -181,9 +183,9 @@ class RemoteAzure(RemoteBase): try: self.blob_service.get_blob_to_path( - bucket, key, tmp_file, progress_callback=cb) + bucket, path, tmp_file, progress_callback=cb) except Exception as exc: - msg = "Failed to download '{}/{}'".format(bucket, key) + msg = "Failed to download '{}/{}'".format(bucket, path) logger.warn(msg, exc) else: os.rename(tmp_file, to_info['path']) @@ -201,8 +203,8 @@ class RemoteAzure(RemoteBase): # 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(key) - for key in self._list_keys(self.bucket, self.prefix) + self._path_to_etag(path) + for path in self._list_keys(self.bucket, self.prefix) ) def gc(self, cinfos): @@ -216,9 +218,9 @@ class RemoteAzure(RemoteBase): if etag in used: continue path_info = {'scheme': self.scheme, - 'key': posixpath.join(self.prefix, - etag[0:2], - etag[2:]), + 'path': posixpath.join(self.prefix, + etag[0:2], + etag[2:]), 'bucket': self.bucket} self.remove(path_info) removed = True diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index 78a9f75eb..f1a3fa7e8 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -48,11 +48,11 @@ class RemoteGS(RemoteBase): def gs(self): return storage.Client() - def get_md5(self, bucket, key): + def get_md5(self, bucket, path): import base64 import codecs - blob = self.gs.bucket(bucket).get_blob(key) + blob = self.gs.bucket(bucket).get_blob(path) if not blob: return None @@ -65,17 +65,17 @@ class RemoteGS(RemoteBase): raise NotImplementedError return {self.PARAM_MD5: self.get_md5(path_info['bucket'], - path_info['key'])} + path_info['path'])} @staticmethod def to_string(path_info): return "{}://{}/{}".format(path_info['scheme'], path_info['bucket'], - path_info['key']) + path_info['path']) def changed_cache(self, md5): - key = posixpath.join(self.prefix, md5[0:2], md5[2:]) - cache = {'scheme': 'gs', 'bucket': self.bucket, 'key': key} + path = posixpath.join(self.prefix, md5[0:2], md5[2:]) + cache = {'scheme': 'gs', 'bucket': self.bucket, 'path': path} if {self.PARAM_MD5: md5} != self.save_info(cache): if self.exists(cache): @@ -102,23 +102,23 @@ class RemoteGS(RemoteBase): def _copy(self, from_info, to_info, gs=None): gs = gs if gs else self.gs - blob = gs.bucket(from_info['bucket']).get_blob(from_info['key']) + blob = gs.bucket(from_info['bucket']).get_blob(from_info['path']) if not blob: msg = '{} doesn\'t exist in the cloud' - raise DvcException(msg.format(from_info['key'])) + raise DvcException(msg.format(from_info['path'])) bucket = self.gs.bucket(to_info['bucket']) bucket.copy_blob(blob, self.gs.bucket(to_info['bucket']), - new_name=to_info['key']) + new_name=to_info['path']) def save(self, path_info): if path_info['scheme'] != 'gs': raise NotImplementedError - md5 = self.get_md5(path_info['bucket'], path_info['key']) - key = posixpath.join(self.prefix, md5[0:2], md5[2:]) - to_info = {'scheme': 'gs', 'bucket': self.bucket, 'key': key} + md5 = self.get_md5(path_info['bucket'], path_info['path']) + path = posixpath.join(self.prefix, md5[0:2], md5[2:]) + to_info = {'scheme': 'gs', 'bucket': self.bucket, 'path': path} self._copy(path_info, to_info) @@ -151,8 +151,8 @@ class RemoteGS(RemoteBase): msg = "Checking out '{}' with cache '{}'." logger.info(msg.format(self.to_string(path_info), md5)) - key = posixpath.join(self.prefix, md5[0:2], md5[2:]) - from_info = {'scheme': 'gs', 'bucket': self.bucket, 'key': key} + path = posixpath.join(self.prefix, md5[0:2], md5[2:]) + from_info = {'scheme': 'gs', 'bucket': self.bucket, 'path': path} self._copy(from_info, path_info) @@ -161,9 +161,9 @@ class RemoteGS(RemoteBase): raise NotImplementedError logger.debug("Removing gs://{}/{}".format(path_info['bucket'], - path_info['key'])) + path_info['path'])) - blob = self.gs.bucket(path_info['bucket']).get_blob(path_info['key']) + blob = self.gs.bucket(path_info['bucket']).get_blob(path_info['path']) if not blob: return @@ -172,10 +172,10 @@ class RemoteGS(RemoteBase): def md5s_to_path_infos(self, md5s): return [{'scheme': 'gs', 'bucket': self.bucket, - 'key': posixpath.join(self.prefix, - md5[0:2], md5[2:])} for md5 in md5s] + 'path': posixpath.join(self.prefix, + md5[0:2], md5[2:])} for md5 in md5s] - def _list_keys(self, bucket, prefix): + def _list_paths(self, bucket, prefix): for blob in self.gs.bucket(bucket).list_blobs(prefix=prefix): yield blob.name @@ -183,8 +183,8 @@ class RemoteGS(RemoteBase): assert not isinstance(path_info, list) assert path_info['scheme'] == 'gs' - keys = self._list_keys(path_info['bucket'], path_info['key']) - return any(path_info['key'] == key for key in keys) + paths = self._list_paths(path_info['bucket'], path_info['path']) + return any(path_info['path'] == path for path in paths) def cache_exists(self, md5s): assert isinstance(md5s, list) @@ -193,13 +193,14 @@ class RemoteGS(RemoteBase): return [] ret = len(md5s) * [False] - keys = [posixpath.join(self.prefix, md5[0:2], md5[2:]) for md5 in md5s] - for key in self._list_keys(self.bucket, self.prefix): - for i, k in enumerate(keys): - if k == key: + paths = [posixpath.join(self.prefix, + md5[0:2], md5[2:]) for md5 in md5s] + for path in self._list_paths(self.bucket, self.prefix): + for i, k in enumerate(paths): + if k == path: ret[i] = True - assert len(ret) == len(keys) == len(md5s) + assert len(ret) == len(paths) == len(md5s) return ret @@ -217,7 +218,7 @@ class RemoteGS(RemoteBase): logger.debug("Uploading '{}' to '{}/{}'".format(from_info['path'], to_info['bucket'], - to_info['key'])) + to_info['path'])) if not name: name = os.path.basename(from_info['path']) @@ -226,13 +227,13 @@ class RemoteGS(RemoteBase): try: bucket = gs.bucket(to_info['bucket']) - blob = bucket.blob(to_info['key']) + blob = bucket.blob(to_info['path']) blob.upload_from_filename(from_info['path']) except Exception as exc: msg = "Failed to upload '{}' to '{}/{}'" logger.warn(msg.format(from_info['path'], to_info['bucket'], - to_info['key']), exc) + to_info['path']), exc) continue progress.finish_target(name) @@ -258,7 +259,7 @@ class RemoteGS(RemoteBase): raise NotImplementedError msg = "Downloading '{}/{}' to '{}'".format(from_info['bucket'], - from_info['key'], + from_info['path'], to_info['path']) logger.debug(msg) @@ -268,19 +269,19 @@ class RemoteGS(RemoteBase): if not no_progress_bar: # percent_cb is not available for download_to_filename, so - # lets at least update progress at keypoints(start, finish) + # lets at least update progress at pathpoints(start, finish) progress.update_target(name, 0, None) self._makedirs(to_info['path']) try: bucket = gs.bucket(from_info['bucket']) - blob = bucket.get_blob(from_info['key']) + blob = bucket.get_blob(from_info['path']) blob.download_to_filename(tmp_file) except Exception as exc: msg = "Failed to download '{}/{}' to '{}'" logger.warn(msg.format(from_info['bucket'], - from_info['key'], + from_info['path'], to_info['path']), exc) continue @@ -299,8 +300,8 @@ class RemoteGS(RemoteBase): # 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(key) - for key in self._list_keys(self.bucket, self.prefix) + self._path_to_md5(path) + for path in self._list_paths(self.bucket, self.prefix) ) def gc(self, cinfos): @@ -313,8 +314,8 @@ class RemoteGS(RemoteBase): continue path_info = {'scheme': 'gs', 'bucket': self.bucket, - 'key': posixpath.join(self.prefix, - md5[0:2], md5[2:])} + 'path': posixpath.join(self.prefix, + md5[0:2], md5[2:])} self.remove(path_info) removed = True diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 633568fb4..0fa9869a8 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -86,11 +86,12 @@ class RemoteS3(RemoteBase): endpoint_url=self.endpoint_url, use_ssl=self.use_ssl) - def get_etag(self, bucket, key): + def get_etag(self, bucket, path): try: - obj = self.s3.head_object(Bucket=bucket, Key=key) + obj = self.s3.head_object(Bucket=bucket, Key=path) except Exception: - raise DvcException('s3://{}/{} does not exist'.format(bucket, key)) + raise DvcException('s3://{}/{} does not exist'.format(bucket, + path)) return obj['ETag'].strip('"') @@ -99,7 +100,7 @@ class RemoteS3(RemoteBase): raise NotImplementedError return {self.PARAM_ETAG: self.get_etag(path_info['bucket'], - path_info['key'])} + path_info['path'])} def changed(self, path_info, checksum_info): if not self.exists(path_info): @@ -118,16 +119,16 @@ class RemoteS3(RemoteBase): s3 = s3 if s3 else self.s3 source = {'Bucket': from_info['bucket'], - 'Key': from_info['key']} - self.s3.copy(source, to_info['bucket'], to_info['key']) + 'Key': from_info['path']} + self.s3.copy(source, to_info['bucket'], to_info['path']) def save(self, path_info): if path_info['scheme'] != 's3': raise NotImplementedError - etag = self.get_etag(path_info['bucket'], path_info['key']) - key = posixpath.join(self.prefix, etag[0:2], etag[2:]) - to_info = {'scheme': 's3', 'bucket': self.bucket, 'key': key} + etag = self.get_etag(path_info['bucket'], path_info['path']) + path = posixpath.join(self.prefix, etag[0:2], etag[2:]) + to_info = {'scheme': 's3', 'bucket': self.bucket, 'path': path} self._copy(path_info, to_info) @@ -135,11 +136,11 @@ class RemoteS3(RemoteBase): @staticmethod def to_string(path_info): - return "s3://{}/{}".format(path_info['bucket'], path_info['key']) + return "s3://{}/{}".format(path_info['bucket'], path_info['path']) def changed_cache(self, etag): - key = posixpath.join(self.prefix, etag[0:2], etag[2:]) - cache = {'scheme': 's3', 'bucket': self.bucket, 'key': key} + path = posixpath.join(self.prefix, etag[0:2], etag[2:]) + cache = {'scheme': 's3', 'bucket': self.bucket, 'path': path} if {self.PARAM_ETAG: etag} != self.save_info(cache): if self.exists(cache): @@ -177,8 +178,8 @@ class RemoteS3(RemoteBase): msg = "Checking out '{}' with cache '{}'." logger.info(msg.format(self.to_string(path_info), etag)) - key = posixpath.join(self.prefix, etag[0:2], etag[2:]) - from_info = {'scheme': 's3', 'bucket': self.bucket, 'key': key} + path = posixpath.join(self.prefix, etag[0:2], etag[2:]) + from_info = {'scheme': 's3', 'bucket': self.bucket, 'path': path} self._copy(from_info, path_info) @@ -187,18 +188,18 @@ class RemoteS3(RemoteBase): raise NotImplementedError logger.debug('Removing s3://{}/{}'.format(path_info['bucket'], - path_info['key'])) + path_info['path'])) self.s3.delete_object(Bucket=path_info['bucket'], - Key=path_info['key']) + Key=path_info['path']) def md5s_to_path_infos(self, md5s): return [{'scheme': self.scheme, 'bucket': self.bucket, - 'key': posixpath.join(self.prefix, - md5[0:2], md5[2:])} for md5 in md5s] + 'path': posixpath.join(self.prefix, + md5[0:2], md5[2:])} for md5 in md5s] - def _list_keys(self, bucket, prefix): + def _list_paths(self, bucket, prefix): s3 = self.s3 kwargs = {'Bucket': bucket, @@ -226,8 +227,8 @@ class RemoteS3(RemoteBase): assert not isinstance(path_info, list) assert path_info['scheme'] == 's3' - keys = self._list_keys(path_info['bucket'], path_info['key']) - return any(path_info['key'] == key for key in keys) + paths = self._list_paths(path_info['bucket'], path_info['path']) + return any(path_info['path'] == path for path in paths) def cache_exists(self, md5s): assert isinstance(md5s, list) @@ -236,13 +237,15 @@ class RemoteS3(RemoteBase): return [] ret = len(md5s) * [False] - keys = [posixpath.join(self.prefix, md5[0:2], md5[2:]) for md5 in md5s] - for key in self._list_keys(self.bucket, self.prefix): - for i, k in enumerate(keys): - if k == key: + paths = [posixpath.join(self.prefix, + md5[0:2], + md5[2:]) for md5 in md5s] + for path in self._list_paths(self.bucket, self.prefix): + for i, k in enumerate(paths): + if k == path: ret[i] = True - assert len(keys) == len(ret) == len(md5s) + assert len(paths) == len(ret) == len(md5s) return ret @@ -260,7 +263,7 @@ class RemoteS3(RemoteBase): logger.debug("Uploading '{}' to '{}/{}'".format(from_info['path'], to_info['bucket'], - to_info['key'])) + to_info['path'])) if not name: name = os.path.basename(from_info['path']) @@ -271,7 +274,7 @@ class RemoteS3(RemoteBase): try: s3.upload_file(from_info['path'], to_info['bucket'], - to_info['key'], + to_info['path'], Callback=cb) except Exception as exc: msg = "Failed to upload '{}'".format(from_info['path']) @@ -301,7 +304,7 @@ class RemoteS3(RemoteBase): raise NotImplementedError msg = "Downloading '{}/{}' to '{}'".format(from_info['bucket'], - from_info['key'], + from_info['path'], to_info['path']) logger.debug(msg) @@ -317,16 +320,16 @@ class RemoteS3(RemoteBase): else: total = s3.head_object( Bucket=from_info['bucket'], - Key=from_info['key'])['ContentLength'] + Key=from_info['path'])['ContentLength'] cb = Callback(name, total) s3.download_file(from_info['bucket'], - from_info['key'], + from_info['path'], tmp_file, Callback=cb) except Exception as exc: msg = "Failed to download '{}/{}'".format(from_info['bucket'], - from_info['key']) + from_info['path']) logger.warn(msg, exc) continue @@ -345,8 +348,8 @@ class RemoteS3(RemoteBase): # 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(key) - for key in self._list_keys(self.bucket, self.prefix) + self._path_to_etag(path) + for path in self._list_paths(self.bucket, self.prefix) ) def gc(self, cinfos): @@ -358,8 +361,8 @@ class RemoteS3(RemoteBase): if etag in used_etags: continue path_info = {'scheme': 's3', - 'key': posixpath.join(self.prefix, - etag[0:2], etag[2:]), + 'path': posixpath.join(self.prefix, + etag[0:2], etag[2:]), 'bucket': self.bucket} self.remove(path_info) removed = True