mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-15 07:18:26 +01:00
remote: use path_infos instead of regular paths in upload/download
Signed-off-by: Ruslan Kuprieiev <[email protected]>
This commit is contained in:
Ruslan Kuprieiev
+27
-21
@@ -152,52 +152,58 @@ class RemoteAzure(RemoteBase):
|
||||
|
||||
return ret
|
||||
|
||||
def upload(self, paths, path_infos, names=None):
|
||||
names = self._verify_path_args(path_infos, paths, names)
|
||||
def upload(self, from_infos, to_infos, names=None):
|
||||
names = self._verify_path_args(to_infos, from_infos, names)
|
||||
|
||||
for path, path_info, name in zip(paths, path_infos, names):
|
||||
if path_info['scheme'] != self.scheme:
|
||||
for from_info, to_info, name in zip(from_infos, to_infos, names):
|
||||
if to_info['scheme'] != self.scheme:
|
||||
raise NotImplementedError
|
||||
|
||||
bucket = path_info['bucket']
|
||||
key = path_info['key']
|
||||
if from_info['scheme'] != 'local':
|
||||
raise NotImplementedError
|
||||
|
||||
bucket = to_info['bucket']
|
||||
key = to_info['key']
|
||||
|
||||
Logger.debug("Uploading '{}' to '{}/{}'".format(
|
||||
path, bucket, key))
|
||||
from_info['path'], bucket, key))
|
||||
|
||||
if not name:
|
||||
name = os.path.basename(path)
|
||||
name = os.path.basename(from_info['path'])
|
||||
|
||||
cb = Callback(name)
|
||||
|
||||
try:
|
||||
self.blob_service.create_blob_from_path(
|
||||
bucket, key, path, progress_callback=cb)
|
||||
bucket, key, from_info['path'], progress_callback=cb)
|
||||
except Exception as ex:
|
||||
Logger.error("Failed to upload '{}'".format(path), ex)
|
||||
Logger.error("Failed to upload '{}'".format(from_info['path']), ex)
|
||||
else:
|
||||
progress.finish_target(name)
|
||||
|
||||
def download(self, path_infos, fnames, no_progress_bar=False, names=None):
|
||||
names = self._verify_path_args(path_infos, fnames, names)
|
||||
def download(self, from_infos, to_infos, no_progress_bar=False, names=None):
|
||||
names = self._verify_path_args(from_infos, to_infos, names)
|
||||
|
||||
for fname, path_info, name in zip(fnames, path_infos, names):
|
||||
if path_info['scheme'] != self.scheme:
|
||||
for to_info, from_info, name in zip(to_infos, from_infos, names):
|
||||
if from_info['scheme'] != self.scheme:
|
||||
raise NotImplementedError
|
||||
|
||||
bucket = path_info['bucket']
|
||||
key = path_info['key']
|
||||
if to_info['scheme'] != 'local':
|
||||
raise NotImplementedError
|
||||
|
||||
bucket = from_info['bucket']
|
||||
key = from_info['key']
|
||||
|
||||
Logger.debug("Downloading '{}/{}' to '{}'".format(
|
||||
bucket, key, fname))
|
||||
bucket, key, to_info['path']))
|
||||
|
||||
tmp_file = self.tmp_file(fname)
|
||||
tmp_file = self.tmp_file(to_info['path'])
|
||||
if not name:
|
||||
name = os.path.basename(fname)
|
||||
name = os.path.basename(to_info['path'])
|
||||
|
||||
cb = None if no_progress_bar else Callback(name)
|
||||
|
||||
self._makedirs(fname)
|
||||
self._makedirs(to_info['path'])
|
||||
|
||||
try:
|
||||
self.blob_service.get_blob_to_path(
|
||||
@@ -206,7 +212,7 @@ class RemoteAzure(RemoteBase):
|
||||
Logger.error("Failed to download '{}/{}'".format(
|
||||
bucket, key), exc)
|
||||
else:
|
||||
os.rename(tmp_file, fname)
|
||||
os.rename(tmp_file, to_info['path'])
|
||||
|
||||
if not no_progress_bar:
|
||||
progress.finish_target(name)
|
||||
|
||||
+8
-8
@@ -72,10 +72,10 @@ class RemoteBase(object):
|
||||
def checkout(self, path_info, checksum_info):
|
||||
raise NotImplementedError
|
||||
|
||||
def download(self, path_info, path, no_progress_bar=False, name=None):
|
||||
def download(self, from_infos, to_infos, no_progress_bar=False, name=None):
|
||||
raise NotImplementedError
|
||||
|
||||
def upload(self, path, path_info, name=None):
|
||||
def upload(self, from_infos, to_infos, path_info, name=None):
|
||||
raise NotImplementedError
|
||||
|
||||
def remove(self, path_info):
|
||||
@@ -99,15 +99,15 @@ class RemoteBase(object):
|
||||
raise NotImplementedError
|
||||
|
||||
@classmethod
|
||||
def _verify_path_args(cls, path_infos, fnames, names=None):
|
||||
assert isinstance(fnames, list)
|
||||
assert isinstance(path_infos, list)
|
||||
assert len(fnames) == len(path_infos)
|
||||
def _verify_path_args(cls, from_infos, to_infos, names=None):
|
||||
assert isinstance(from_infos, list)
|
||||
assert isinstance(to_infos, list)
|
||||
assert len(from_infos) == len(to_infos)
|
||||
|
||||
if not names:
|
||||
names = len(fnames) * [None]
|
||||
names = len(to_infos) * [None]
|
||||
else:
|
||||
assert isinstance(names, list)
|
||||
assert len(names) == len(fnames)
|
||||
assert len(names) == len(to_infos)
|
||||
|
||||
return names
|
||||
|
||||
+33
-27
@@ -116,67 +116,73 @@ class RemoteGS(RemoteBase):
|
||||
ret.append(exists)
|
||||
return ret
|
||||
|
||||
def upload(self, paths, path_infos, names=None):
|
||||
names = self._verify_path_args(path_infos, paths, names)
|
||||
def upload(self, from_infos, to_infos, names=None):
|
||||
names = self._verify_path_args(to_infos, from_infos, names)
|
||||
|
||||
gs = self.gs
|
||||
|
||||
for path, path_info, name in zip(paths, path_infos, names):
|
||||
if path_info['scheme'] != 'gs':
|
||||
for from_info, to_info, name in zip(from_infos, to_infos, names):
|
||||
if to_info['scheme'] != 'gs':
|
||||
raise NotImplementedError
|
||||
|
||||
Logger.debug("Uploading '{}' to '{}/{}'".format(path,
|
||||
path_info['bucket'],
|
||||
path_info['key']))
|
||||
if from_info['scheme'] != 'local':
|
||||
raise NotImplementedError
|
||||
|
||||
Logger.debug("Uploading '{}' to '{}/{}'".format(from_info['path'],
|
||||
to_info['bucket'],
|
||||
to_info['key']))
|
||||
|
||||
if not name:
|
||||
name = os.path.basename(path)
|
||||
name = os.path.basename(from_info['path'])
|
||||
|
||||
progress.update_target(name, 0, None)
|
||||
|
||||
try:
|
||||
gs.bucket(path_info['bucket']).blob(path_info['key']).upload_from_filename(path)
|
||||
gs.bucket(to_info['bucket']).blob(to_info['key']).upload_from_filename(from_info['path'])
|
||||
except Exception as exc:
|
||||
Logger.error("Failed to upload '{}' to '{}/{}'".format(path,
|
||||
path_info['bucket'],
|
||||
path_info['key']), exc)
|
||||
Logger.error("Failed to upload '{}' to '{}/{}'".format(from_info['path'],
|
||||
to_info['bucket'],
|
||||
to_info['key']), exc)
|
||||
continue
|
||||
|
||||
progress.finish_target(name)
|
||||
|
||||
def download(self, path_infos, paths, no_progress_bar=False, names=None):
|
||||
names = self._verify_path_args(path_infos, paths, names)
|
||||
def download(self, from_infos, to_infos, no_progress_bar=False, names=None):
|
||||
names = self._verify_path_args(from_infos, to_infos, names)
|
||||
|
||||
gs = self.gs
|
||||
|
||||
for path, path_info, name in zip(paths, path_infos, names):
|
||||
if path_info['scheme'] != 'gs':
|
||||
for to_info, from_info, name in zip(to_infos, from_infos, names):
|
||||
if from_info['scheme'] != 'gs':
|
||||
raise NotImplementedError
|
||||
|
||||
Logger.debug("Downloading '{}/{}' to '{}'".format(path_info['bucket'],
|
||||
path_info['key'],
|
||||
path))
|
||||
if to_info['scheme'] != 'local':
|
||||
raise NotImplementedError
|
||||
|
||||
tmp_file = self.tmp_file(path)
|
||||
Logger.debug("Downloading '{}/{}' to '{}'".format(from_info['bucket'],
|
||||
from_info['key'],
|
||||
to_info['path']))
|
||||
|
||||
tmp_file = self.tmp_file(to_info['path'])
|
||||
if not name:
|
||||
name = os.path.basename(path)
|
||||
name = os.path.basename(to_info['path'])
|
||||
|
||||
if not no_progress_bar:
|
||||
# percent_cb is not available for download_to_filename, so
|
||||
# lets at least update progress at keypoints(start, finish)
|
||||
progress.update_target(name, 0, None)
|
||||
|
||||
self._makedirs(path)
|
||||
self._makedirs(to_info['path'])
|
||||
|
||||
try:
|
||||
gs.bucket(path_info['bucket']).get_blob(path_info['key']).download_to_filename(tmp_file)
|
||||
gs.bucket(from_info['bucket']).get_blob(from_info['key']).download_to_filename(tmp_file)
|
||||
except Exception as exc:
|
||||
Logger.error("Failed to download '{}/{}' to '{}'".format(path_info['bucket'],
|
||||
path_info['key'],
|
||||
path), exc)
|
||||
Logger.error("Failed to download '{}/{}' to '{}'".format(from_info['bucket'],
|
||||
from_info['key'],
|
||||
to_info['path']), exc)
|
||||
continue
|
||||
|
||||
os.rename(tmp_file, path)
|
||||
os.rename(tmp_file, to_info['path'])
|
||||
|
||||
if not no_progress_bar:
|
||||
progress.finish_target(name)
|
||||
|
||||
+20
-14
@@ -136,28 +136,34 @@ class RemoteHDFS(RemoteBase):
|
||||
|
||||
return ret
|
||||
|
||||
def upload(self, paths, path_infos, names=None):
|
||||
names = self._verify_path_args(path_infos, paths, names)
|
||||
def upload(self, from_infos, to_infos, names=None):
|
||||
names = self._verify_path_args(to_infos, from_infos, names)
|
||||
|
||||
for path, path_info, name in zip(paths, path_infos, names):
|
||||
if path_info['scheme'] != 'hdfs':
|
||||
for from_info, to_info, name in zip(from_infos, to_infos, names):
|
||||
if to_info['scheme'] != 'hdfs':
|
||||
raise NotImplementedError
|
||||
|
||||
self.hadoop_fs('mkdir -p {}'.format(posixpath.dirname(path_info['url'])), user=path_info['user'])
|
||||
self.hadoop_fs('copyFromLocal {} {}'.format(path, path_info['url']), user=path_info['user'])
|
||||
|
||||
def download(self, path_infos, paths, no_progress_bar=False, names=None):
|
||||
names = self._verify_path_args(path_infos, paths, names)
|
||||
|
||||
for path, path_info, name in zip(paths, path_infos, names):
|
||||
if path_info['scheme'] != 'hdfs':
|
||||
if from_info['scheme'] != 'local':
|
||||
raise NotImplementedError
|
||||
|
||||
dname = os.path.dirname(path)
|
||||
self.hadoop_fs('mkdir -p {}'.format(posixpath.dirname(to_info['url'])), user=to_info['user'])
|
||||
self.hadoop_fs('copyFromLocal {} {}'.format(from_info['path'], to_info['url']), user=to_info['user'])
|
||||
|
||||
def download(self, from_infos, to_infos, no_progress_bar=False, names=None):
|
||||
names = self._verify_path_args(from_infos, to_infos, names)
|
||||
|
||||
for to_info, from_info, name in zip(to_infos, from_infos, names):
|
||||
if from_info['scheme'] != 'hdfs':
|
||||
raise NotImplementedError
|
||||
|
||||
if to_info['scheme'] != 'local':
|
||||
raise NotImplementedError
|
||||
|
||||
dname = os.path.dirname(to_info['path'])
|
||||
if not os.path.exists(dname):
|
||||
os.makedirs(dname)
|
||||
|
||||
self.hadoop_fs('copyToLocal {} {}'.format(path_info['url'], path), user=path_info['user'])
|
||||
self.hadoop_fs('copyToLocal {} {}'.format(from_info['url'], to_info['path']), user=from_info['user'])
|
||||
|
||||
def _path_to_checksum(self, path):
|
||||
relpath = posixpath.relpath(path, self.url)
|
||||
|
||||
+41
-35
@@ -270,46 +270,52 @@ class RemoteLOCAL(RemoteBase):
|
||||
ret.append(os.path.exists(path_info['path']))
|
||||
return ret
|
||||
|
||||
def upload(self, paths, path_infos, names=None):
|
||||
names = self._verify_path_args(path_infos, paths, names)
|
||||
def upload(self, from_infos, to_infos, names=None):
|
||||
names = self._verify_path_args(to_infos, from_infos, names)
|
||||
|
||||
for path, path_info, name in zip(paths, path_infos, names):
|
||||
if path_info['scheme'] != 'local':
|
||||
for from_info, to_info, name in zip(from_infos, to_infos, names):
|
||||
if to_info['scheme'] != 'local':
|
||||
raise NotImplementedError
|
||||
|
||||
Logger.debug("Uploading '{}' to '{}'".format(path, path_info['path']))
|
||||
|
||||
if not name:
|
||||
name = os.path.basename(path)
|
||||
|
||||
self._makedirs(path_info['path'])
|
||||
|
||||
try:
|
||||
copyfile(path, path_info['path'], name=name)
|
||||
except Exception as exc:
|
||||
Logger.error("Failed to upload '{}' tp '{}'".format(path, path_info['path']))
|
||||
|
||||
def download(self, path_infos, paths, no_progress_bar=False, names=None):
|
||||
names = self._verify_path_args(path_infos, paths, names)
|
||||
|
||||
for path, path_info, name in zip(paths, path_infos, names):
|
||||
if path_info['scheme'] != 'local':
|
||||
if from_info['scheme'] != 'local':
|
||||
raise NotImplementedError
|
||||
|
||||
Logger.debug("Downloading '{}' to '{}'".format(path_info['path'], path))
|
||||
Logger.debug("Uploading '{}' to '{}'".format(from_info['path'], to_info['path']))
|
||||
|
||||
if not name:
|
||||
name = os.path.basename(path)
|
||||
name = os.path.basename(from_info['path'])
|
||||
|
||||
self._makedirs(to_info['path'])
|
||||
|
||||
self._makedirs(path)
|
||||
tmp_file = self.tmp_file(path)
|
||||
try:
|
||||
copyfile(path_info['path'], tmp_file, no_progress_bar=no_progress_bar, name=name)
|
||||
copyfile(from_info['path'], to_info['path'], name=name)
|
||||
except Exception as exc:
|
||||
Logger.error("Failed to download '{}' to '{}'".format(path_info['path'], path), exc)
|
||||
Logger.error("Failed to upload '{}' tp '{}'".format(from_info['path'], to_info['path']), exc)
|
||||
|
||||
def download(self, from_infos, to_infos, no_progress_bar=False, names=None):
|
||||
names = self._verify_path_args(from_infos, to_infos, names)
|
||||
|
||||
for to_info, from_info, name in zip(to_infos, from_infos, names):
|
||||
if from_info['scheme'] != 'local':
|
||||
raise NotImplementedError
|
||||
|
||||
if to_info['scheme'] != 'local':
|
||||
raise NotImplementedError
|
||||
|
||||
Logger.debug("Downloading '{}' to '{}'".format(from_info['path'], to_info['path']))
|
||||
|
||||
if not name:
|
||||
name = os.path.basename(to_info['path'])
|
||||
|
||||
self._makedirs(to_info['path'])
|
||||
tmp_file = self.tmp_file(to_info['path'])
|
||||
try:
|
||||
copyfile(from_info['path'], tmp_file, no_progress_bar=no_progress_bar, name=name)
|
||||
except Exception as exc:
|
||||
Logger.error("Failed to download '{}' to '{}'".format(from_info['path'], to_info['path']), exc)
|
||||
continue
|
||||
|
||||
os.rename(tmp_file, path)
|
||||
os.rename(tmp_file, to_info['path'])
|
||||
|
||||
def _collect(self, checksum_infos):
|
||||
missing = []
|
||||
@@ -349,7 +355,7 @@ class RemoteLOCAL(RemoteBase):
|
||||
# NOTE: filter files that are not corrupted
|
||||
md5s = list(filter(lambda md5: self.changed(md5), md5s))
|
||||
|
||||
cache = [self.get(md5) for md5 in md5s]
|
||||
cache = [{'scheme': 'local', 'path': self.get(md5)} for md5 in md5s]
|
||||
path_infos = remote.md5s_to_path_infos(md5s)
|
||||
|
||||
assert len(path_infos) == len(cache) == len(md5s)
|
||||
@@ -364,9 +370,9 @@ class RemoteLOCAL(RemoteBase):
|
||||
return
|
||||
|
||||
with ThreadPoolExecutor(max_workers=len(chunks)) as executor:
|
||||
for path_infos, paths, md5s in chunks:
|
||||
executor.submit(remote.download, path_infos,
|
||||
paths,
|
||||
for from_infos, to_infos, md5s in chunks:
|
||||
executor.submit(remote.download, from_infos,
|
||||
to_infos,
|
||||
names=md5s,
|
||||
no_progress_bar=no_progress_bar)
|
||||
|
||||
@@ -390,7 +396,7 @@ class RemoteLOCAL(RemoteBase):
|
||||
md5s_exist = filter(lambda x: not x[1], list(zip(md5s, remote.exists(path_infos))))
|
||||
md5s = [md5 for md5, exists in md5s_exist]
|
||||
|
||||
cache = [self.get(md5) for md5 in md5s]
|
||||
cache = [{'scheme': 'local', 'path': self.get(md5)} for md5 in md5s]
|
||||
path_infos = remote.md5s_to_path_infos(md5s)
|
||||
|
||||
assert len(path_infos) == len(cache) == len(md5s)
|
||||
@@ -405,5 +411,5 @@ class RemoteLOCAL(RemoteBase):
|
||||
return
|
||||
|
||||
with ThreadPoolExecutor(max_workers=len(chunks)) as executor:
|
||||
for path_infos, paths, md5s in chunks:
|
||||
executor.submit(remote.upload, paths, path_infos, names=md5s)
|
||||
for to_infos, from_infos, md5s in chunks:
|
||||
executor.submit(remote.upload, from_infos, to_infos, names=md5s)
|
||||
|
||||
+33
-27
@@ -156,69 +156,75 @@ class RemoteS3(RemoteBase):
|
||||
|
||||
return ret
|
||||
|
||||
def upload(self, paths, path_infos, names=None):
|
||||
names = self._verify_path_args(path_infos, paths, names)
|
||||
def upload(self, from_infos, path_infos, names=None):
|
||||
names = self._verify_path_args(to_infos, from_infos, names)
|
||||
|
||||
session = boto3.session.Session()
|
||||
s3 = session.client('s3')
|
||||
|
||||
for path, path_info, name in zip(paths, path_infos, names):
|
||||
if path_info['scheme'] != 's3':
|
||||
for from_info, to_info, name in zip(from_infos, to_infos, names):
|
||||
if to_info['scheme'] != 's3':
|
||||
raise NotImplementedError
|
||||
|
||||
Logger.debug("Uploading '{}' to '{}/{}'".format(path,
|
||||
path_info['bucket'],
|
||||
path_info['key']))
|
||||
if from_info['scheme'] != 'local':
|
||||
raise NotImplementedError
|
||||
|
||||
Logger.debug("Uploading '{}' to '{}/{}'".format(from_info['path'],
|
||||
to_info['bucket'],
|
||||
to_info['key']))
|
||||
|
||||
if not name:
|
||||
name = os.path.basename(path)
|
||||
name = os.path.basename(from_info['path'])
|
||||
|
||||
total = os.path.getsize(path)
|
||||
total = os.path.getsize(from_info['path'])
|
||||
cb = Callback(name, total)
|
||||
|
||||
try:
|
||||
s3.upload_file(path, path_info['bucket'], path_info['key'], Callback=cb)
|
||||
s3.upload_file(from_info['path'], to_info['bucket'], to_info['key'], Callback=cb)
|
||||
except Exception as exc:
|
||||
Logger.error("Failed to upload '{}'".format(path), exc)
|
||||
Logger.error("Failed to upload '{}'".format(from_info['path']), exc)
|
||||
continue
|
||||
|
||||
progress.finish_target(name)
|
||||
|
||||
def download(self, path_infos, fnames, no_progress_bar=False, names=None):
|
||||
names = self._verify_path_args(path_infos, fnames, names)
|
||||
def download(self, from_infos, to_infos, no_progress_bar=False, names=None):
|
||||
names = self._verify_path_args(from_infos, to_infos, names)
|
||||
|
||||
session = boto3.session.Session()
|
||||
s3 = session.client('s3')
|
||||
|
||||
for fname, path_info, name in zip(fnames, path_infos, names):
|
||||
if path_info['scheme'] != 's3':
|
||||
for to_info, from_info, name in zip(to_infos, from_infos, names):
|
||||
if from_info['scheme'] != 's3':
|
||||
raise NotImplementedError
|
||||
|
||||
Logger.debug("Downloading '{}/{}' to '{}'".format(path_info['bucket'],
|
||||
path_info['key'],
|
||||
fname))
|
||||
if to_info['scheme'] != 'local':
|
||||
raise NotImplementedError
|
||||
|
||||
tmp_file = self.tmp_file(fname)
|
||||
Logger.debug("Downloading '{}/{}' to '{}'".format(from_info['bucket'],
|
||||
from_info['key'],
|
||||
to_info['path']))
|
||||
|
||||
tmp_file = self.tmp_file(to_info['path'])
|
||||
if not name:
|
||||
name = os.path.basename(fname)
|
||||
name = os.path.basename(to_info['path'])
|
||||
|
||||
if no_progress_bar:
|
||||
cb = None
|
||||
else:
|
||||
total = s3.head_object(Bucket=path_info['bucket'],
|
||||
Key=path_info['key'])['ContentLength']
|
||||
total = s3.head_object(Bucket=from_info['bucket'],
|
||||
Key=from_info['key'])['ContentLength']
|
||||
cb = Callback(name, total)
|
||||
|
||||
self._makedirs(fname)
|
||||
self._makedirs(to_info['path'])
|
||||
|
||||
try:
|
||||
s3.download_file(path_info['bucket'], path_info['key'], tmp_file, Callback=cb)
|
||||
s3.download_file(from_info['bucket'], from_info['key'], tmp_file, Callback=cb)
|
||||
except Exception as exc:
|
||||
Logger.error("Failed to download '{}/{}'".format(path_info['bucket'],
|
||||
path_info['key']), exc)
|
||||
Logger.error("Failed to download '{}/{}'".format(from_info['bucket'],
|
||||
from_info['key']), exc)
|
||||
return
|
||||
|
||||
os.rename(tmp_file, fname)
|
||||
os.rename(tmp_file, to_info['path'])
|
||||
|
||||
if not no_progress_bar:
|
||||
progress.finish_target(name)
|
||||
|
||||
+37
-31
@@ -180,62 +180,68 @@ class RemoteSSH(RemoteBase):
|
||||
ssh.open_sftp().remove(path_info['path'])
|
||||
ssh.close()
|
||||
|
||||
def download(self, path_infos, paths, no_progress_bar=False, names=None):
|
||||
names = self._verify_path_args(path_infos, paths, names)
|
||||
def download(self, from_infos, to_infos, no_progress_bar=False, names=None):
|
||||
names = self._verify_path_args(from_infos, to_infos, names)
|
||||
|
||||
ssh = self.ssh(host=path_infos[0]['host'], user=path_infos[0]['user'])
|
||||
ssh = self.ssh(host=from_infos[0]['host'], user=from_infos[0]['user'])
|
||||
|
||||
for path, path_info, name in zip(paths, path_infos, names):
|
||||
if path_info['scheme'] != 'ssh':
|
||||
for to_info, from_info, name in zip(to_infos, from_infos, names):
|
||||
if from_info['scheme'] != 'ssh':
|
||||
raise NotImplementedError
|
||||
|
||||
Logger.debug("Downloading '{}/{}' to '{}'".format(path_info['host'],
|
||||
path_info['path'],
|
||||
path))
|
||||
if not name:
|
||||
name = os.path.basename(path)
|
||||
if to_info['scheme'] != 'local':
|
||||
raise NotImplementedError
|
||||
|
||||
self._makedirs(path)
|
||||
tmp_file = self.tmp_file(path)
|
||||
Logger.debug("Downloading '{}/{}' to '{}'".format(from_info['host'],
|
||||
from_info['path'],
|
||||
to_info['path']))
|
||||
if not name:
|
||||
name = os.path.basename(to_info['path'])
|
||||
|
||||
self._makedirs(to_info['path'])
|
||||
tmp_file = self.tmp_file(to_info['path'])
|
||||
try:
|
||||
ssh.open_sftp().get(path_info['path'], tmp_file, callback=create_cb(name))
|
||||
ssh.open_sftp().get(from_info['path'], tmp_file, callback=create_cb(name))
|
||||
except Exception as exc:
|
||||
Logger.error("Failed to download '{}/{}' to '{}'".format(path_info['host'],
|
||||
path_info['path'],
|
||||
path), exc)
|
||||
Logger.error("Failed to download '{}/{}' to '{}'".format(from_info['host'],
|
||||
from_info['path'],
|
||||
to_info['path']), exc)
|
||||
continue
|
||||
|
||||
os.rename(tmp_file, path)
|
||||
os.rename(tmp_file, to_info['path'])
|
||||
progress.finish_target(name)
|
||||
|
||||
ssh.close()
|
||||
|
||||
def upload(self, paths, path_infos, names=None):
|
||||
names = self._verify_path_args(path_infos, paths, names)
|
||||
def upload(self, from_infos, to_infos, names=None):
|
||||
names = self._verify_path_args(to_infos, from_infos, names)
|
||||
|
||||
ssh = self.ssh(host=path_infos[0]['host'], user=path_infos[0]['user'])
|
||||
ssh = self.ssh(host=to_infos[0]['host'], user=to_infos[0]['user'])
|
||||
sftp = ssh.open_sftp()
|
||||
|
||||
for path, path_info, name in zip(paths, path_infos, names):
|
||||
if path_info['scheme'] != 'ssh':
|
||||
for from_info, to_info, name in zip(from_infos, to_infos, names):
|
||||
if to_info['scheme'] != 'ssh':
|
||||
raise NotImplementedError
|
||||
|
||||
Logger.debug("Uploading '{}' to '{}/{}'".format(path,
|
||||
path_info['host'],
|
||||
path_info['path']))
|
||||
if from_info['scheme'] != 'local':
|
||||
raise NotImplementedError
|
||||
|
||||
Logger.debug("Uploading '{}' to '{}/{}'".format(from_info['path'],
|
||||
to_info['host'],
|
||||
to_info['path']))
|
||||
|
||||
if not name:
|
||||
name = os.path.basename(path)
|
||||
name = os.path.basename(from_info['path'])
|
||||
|
||||
dname = posixpath.dirname(path_info['path'])
|
||||
dname = posixpath.dirname(to_info['path'])
|
||||
self._exec(ssh, 'mkdir -p {}'.format(dname))
|
||||
|
||||
try:
|
||||
sftp.put(path, path_info['path'], callback=create_cb(name))
|
||||
sftp.put(from_info['path'], to_info['path'], callback=create_cb(name))
|
||||
except Exception as exc:
|
||||
Logger.error("Failed to upload '{}' to '{}/{}'".format(path,
|
||||
path_info['host'],
|
||||
path_info['path']), exc)
|
||||
Logger.error("Failed to upload '{}' to '{}/{}'".format(from_info['path'],
|
||||
to_info['host'],
|
||||
to_info['path']), exc)
|
||||
continue
|
||||
|
||||
progress.finish_target(name)
|
||||
|
||||
+1
-1
@@ -282,7 +282,7 @@ class Stage(object):
|
||||
msg = u'Importing \'{}\' -> \'{}\''
|
||||
self.project.logger.info(msg.format(self.deps[0].path, self.outs[0].rel_path))
|
||||
|
||||
self.deps[0].download(self.outs[0].path)
|
||||
self.deps[0].download(self.outs[0].path_info)
|
||||
elif self.is_data_source:
|
||||
self.project.logger.info(u'Verifying data sources in \'{}\''.format(self.relpath))
|
||||
self.check_missing_outputs()
|
||||
|
||||
Reference in New Issue
Block a user