diff --git a/dvc/command/imp.py b/dvc/command/imp.py index f5119e7be..abe497838 100644 --- a/dvc/command/imp.py +++ b/dvc/command/imp.py @@ -7,6 +7,6 @@ class CmdImport(CmdBase): try: self.project.imp(self.args.url, self.args.out) except DvcException as ex: - self.project.logger.error('Failed to import {}', ex) + self.project.logger.error('Failed to import {}'.format(self.args.url), ex) return 1 return 0 diff --git a/dvc/dependency/base.py b/dvc/dependency/base.py index 9b887e765..1fef1a63a 100644 --- a/dvc/dependency/base.py +++ b/dvc/dependency/base.py @@ -49,6 +49,10 @@ class DependencyBase(object): def sep(self): return '/' + @property + def exists(self): + return self.remote.exists([self.path_info]) + def changed(self): raise NotImplementedError diff --git a/dvc/dependency/local.py b/dvc/dependency/local.py index 47a903929..20705f75d 100644 --- a/dvc/dependency/local.py +++ b/dvc/dependency/local.py @@ -47,7 +47,7 @@ class DependencyLOCAL(DependencyBase): return os.path.relpath(self.path) def changed(self): - if not os.path.exists(self.path): + if not self.exists: return True info = self.remote.save_info(self.path_info) @@ -55,7 +55,7 @@ class DependencyLOCAL(DependencyBase): return self.info != info def save(self): - if not os.path.exists(self.path): + if not self.exists: raise self.DoesNotExistError(self.rel_path) if not os.path.isfile(self.path) and not os.path.isdir(self.path): diff --git a/dvc/project.py b/dvc/project.py index 66c0d8366..ec4945ece 100644 --- a/dvc/project.py +++ b/dvc/project.py @@ -211,14 +211,10 @@ class Project(object): return stage def imp(self, url, out): - stage_fname = out + Stage.STAGE_FILE_SUFFIX - cwd = os.path.dirname(os.path.abspath(out)) stage = Stage.loads(project=self, cmd=None, deps=[url], - outs=[out], - fname=stage_fname, - cwd=cwd) + outs=[out]) stage.run() stage.dump() diff --git a/dvc/remote/azure.py b/dvc/remote/azure.py index a8633e5fa..daca6d0bb 100644 --- a/dvc/remote/azure.py +++ b/dvc/remote/azure.py @@ -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) diff --git a/dvc/remote/base.py b/dvc/remote/base.py index da101a808..0a6b6b4af 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -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 diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index 2b951280b..aa2b43500 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -55,18 +55,24 @@ class RemoteGS(RemoteBase): return {self.PARAM_ETAG: self.get_etag(path_info['bucket'], path_info['key'])} + def _copy(from_info, to_info, gs=None): + gs = gs if gs else self.gs + + blob = gs.bucket(from_info['bucket']).get_blob(from_info['key']) + if not blob: + raise DvcException('{} doesn\'t exist in the cloud'.format(from_info['key'])) + + self.gs.bucket(to_info['bucket']).copy_blob(blob, self.gs.bucket(to_info['bucket']), new_name=to_info['key']) + def save(self, path_info): if path_info['scheme'] != 'gs': raise NotImplementedError etag = self.get_etag(path_info['bucket'], path_info['key']) - dest_key = posixpath.join(self.prefix, etag[0:2], etag[2:]) + key = posixpath.join(self.prefix, etag[0:2], etag[2:]) + to_info = {'scheme': 'gs', 'bucket': self.bucket, 'key': key} - blob = self.gs.bucket(path_info['bucket']).get_blob(path_info['key']) - if not blob: - raise DvcException('{} doesn\'t exist in the cloud'.format(path_info['key'])) - - self.gs.bucket(self.bucket).copy_blob(blob, self.gs.bucket(path_info['bucket']), new_name=dest_key) + self._copy(path_info, to_info) return {self.PARAM_ETAG: etag} @@ -79,11 +85,9 @@ class RemoteGS(RemoteBase): return key = posixpath.join(self.prefix, etag[0:2], etag[2:]) - blob = self.gs.bucket(self.bucket).get_blob(key) - if not blob: - raise DvcException('{} doesn\'t exist in the cloud'.format(key)) + from_info = {'scheme': 'gs', 'bucket': self.bucket, 'key': key} - self.gs.bucket(path_info['bucket']).copy_blob(blob, self.gs.bucket(self.bucket), new_name=path_info['key']) + self._copy(from_info, path_info) def remove(self, path_info): if path_info['scheme'] != 'gs': @@ -116,67 +120,77 @@ 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 from_info['scheme'] == 'gs': + self._copy(from_info, to_info, gs=gs) + continue - tmp_file = self.tmp_file(path) + if to_info['scheme'] != 'local': + raise NotImplementedError + + 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) diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py index 22f67bea6..9c73c0b0e 100644 --- a/dvc/remote/hdfs.py +++ b/dvc/remote/hdfs.py @@ -57,7 +57,7 @@ class RemoteHDFS(RemoteBase): 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 {} {}'.format(from_info['url'], to_info['url']), user=to_info['user']) + self.hadoop_fs('cp -f {} {}'.format(from_info['url'], to_info['url']), user=to_info['user']) def rm(self, path_info): self.hadoop_fs('rm {}'.format(path_info['url']), user=path_info['user']) @@ -136,28 +136,38 @@ 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'] == 'hdfs': + self.cp(from_info, to_info) + continue + + 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) diff --git a/dvc/remote/local.py b/dvc/remote/local.py index bac674cf1..a351181b8 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -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) diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 946f3b4b6..a6ea8a863 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -18,6 +18,7 @@ from dvc.progress import progress from dvc.config import Config from dvc.remote.base import RemoteBase from dvc.remote.local import RemoteLOCAL +from dvc.exceptions import DvcException class Callback(object): @@ -70,7 +71,7 @@ class RemoteS3(RemoteBase): try: obj = self.s3.Object(bucket, key).get() except Exception: - return None + raise DvcException('s3://{}/{} does not exist'.format(bucket, key)) return obj['ETag'].strip('"') @@ -80,16 +81,22 @@ class RemoteS3(RemoteBase): return {self.PARAM_ETAG: self.get_etag(path_info['bucket'], path_info['key'])} + def _copy(self, from_info, to_info, s3=None): + s3 = s3 if s3 else self.s3 + + source = {'Bucket': from_info['bucket'], + 'Key': from_info['key']} + self.s3.Bucket(to_info['bucket']).copy(source, to_info['key']) + def save(self, path_info): if path_info['scheme'] != 's3': raise NotImplementedError etag = self.get_etag(path_info['bucket'], path_info['key']) - dest_key = posixpath.join(self.prefix, etag[0:2], etag[2:]) + key = posixpath.join(self.prefix, etag[0:2], etag[2:]) + to_info = {'scheme': 's3', 'bucket': self.bucket, 'key': key} - source = {'Bucket': path_info['bucket'], - 'Key': path_info['key']} - self.s3.Bucket(self.bucket).copy(source, dest_key) + self._copy(path_info, to_info) return {self.PARAM_ETAG: etag} @@ -102,10 +109,9 @@ class RemoteS3(RemoteBase): return key = posixpath.join(self.prefix, etag[0:2], etag[2:]) - source = {'Bucket': self.bucket, - 'Key': key} + from_info = {'scheme': 's3', 'bucket': self.bucket, 'key': key} - self.s3.Bucket(path_info['bucket']).copy(source, path_info['key']) + self._copy(from_info, path_info) def remove(self, path_info): if path_info['scheme'] != 's3': @@ -161,67 +167,77 @@ 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) s3 = self.s3_session_client - 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) s3 = self.s3_session_client - 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'] == 's3': + self._copy(from_info, to_info, s3=s3) + continue - tmp_file = self.tmp_file(fname) + if to_info['scheme'] != 'local': + raise NotImplementedError + + 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) diff --git a/dvc/remote/ssh.py b/dvc/remote/ssh.py index af1c1684a..c64fd57e3 100644 --- a/dvc/remote/ssh.py +++ b/dvc/remote/ssh.py @@ -122,20 +122,21 @@ class RemoteSSH(RemoteBase): return md5 - def cp(self, from_info, to_info): + def cp(self, from_info, to_info, ssh=None): if from_info['scheme'] != 'ssh' or to_info['scheme'] != 'ssh': raise NotImplementedError assert from_info['host'] == to_info['host'] assert from_info['user'] == to_info['user'] - ssh = self.ssh(host=from_info['host'], user=from_info['user']) + s = ssh if ssh else self.ssh(host=from_info['host'], user=from_info['user']) dname = posixpath.dirname(to_info['path']) - self._exec(ssh, 'mkdir -p {}'.format(dname)) - self._exec(ssh, 'cp {} {}'.format(from_info['path'], to_info['path'])) + self._exec(s, 'mkdir -p {}'.format(dname)) + self._exec(s, 'cp {} {}'.format(from_info['path'], to_info['path'])) - ssh.close() + if not ssh: + s.close() def save_info(self, path_info): if path_info['scheme'] != 'ssh': @@ -180,62 +181,74 @@ 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) - - self._makedirs(path) - tmp_file = self.tmp_file(path) - try: - ssh.open_sftp().get(path_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) + if to_info['scheme'] == 'ssh': + assert from_info['host'] == to_info['host'] + assert from_info['user'] == to_info['user'] + self.cp(from_info, to_info, ssh=ssh) continue - os.rename(tmp_file, path) + if to_info['scheme'] != 'local': + raise NotImplementedError + + 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(from_info['path'], tmp_file, callback=create_cb(name)) + except Exception as exc: + Logger.error("Failed to download '{}/{}' to '{}'".format(from_info['host'], + from_info['path'], + to_info['path']), exc) + continue + + 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) diff --git a/dvc/stage.py b/dvc/stage.py index 7fcbce268..8c247273d 100644 --- a/dvc/stage.py +++ b/dvc/stage.py @@ -103,8 +103,7 @@ class Stage(object): def is_import(self): return not self.cmd and \ len(self.deps) == 1 and \ - len(self.outs) == 1 and \ - self.outs[0].path_info['scheme'] == 'local' + len(self.outs) == 1 def changed(self): ret = False @@ -280,9 +279,9 @@ class Stage(object): self.check_missing_outputs() elif self.is_import: msg = u'Importing \'{}\' -> \'{}\'' - self.project.logger.info(msg.format(self.deps[0].path, self.outs[0].rel_path)) + self.project.logger.info(msg.format(self.deps[0].path, self.outs[0].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() @@ -301,9 +300,10 @@ class Stage(object): self.save() def check_missing_outputs(self): - missing_outs = [out.rel_path for out in self.outs if not os.path.exists(out.rel_path)] - if missing_outs: - raise MissingDataSource(missing_outs) + outs = [out for out in self.outs if not out.exists] + paths = [out.path if out.path_info['scheme'] != 'local' else out.rel_path for out in outs] + if paths: + raise MissingDataSource(paths) def checkout(self): for out in self.outs: diff --git a/tests/test_repro.py b/tests/test_repro.py index e88819960..5fa913723 100644 --- a/tests/test_repro.py +++ b/tests/test_repro.py @@ -337,6 +337,10 @@ class TestReproExternalBase(TestDvc): self.assertTrue(os.path.exists('import')) self.assertTrue(filecmp.cmp('import', self.FOO)) + import_remote_stage = self.dvc.imp(out_foo_path, out_foo_path + '_imported') + + add_stage = self.dvc.add(out_foo_path) + cmd_stage = self.dvc.run(outs=[out_bar_path], deps=[out_foo_path], cmd=self.cmd(foo_path, bar_path)) @@ -346,7 +350,7 @@ class TestReproExternalBase(TestDvc): sleep() stages = self.dvc.reproduce(import_stage.path) - self.assertEqual(len(stages), 1) + self.assertEqual(len(stages), 2) self.assertTrue(os.path.exists('import')) self.assertTrue(filecmp.cmp('import', self.BAR))