mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-11 07:18:28 +01:00
remote: RFC: use 'path' instead of 'key'
Signed-off-by: Ruslan Kuprieiev <[email protected]>
This commit is contained in:
Ruslan Kuprieiev
@@ -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}
|
||||
|
||||
@@ -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:
|
||||
|
||||
+23
-21
@@ -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
|
||||
|
||||
+38
-37
@@ -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
|
||||
|
||||
|
||||
+39
-36
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user