diff --git a/dvc/command/import_file.py b/dvc/command/import_file.py index 55e552633..2ff486398 100644 --- a/dvc/command/import_file.py +++ b/dvc/command/import_file.py @@ -107,9 +107,8 @@ class CmdImportFile(CmdBase): raise ImportFileError('Output file directory "{}" does not exists'.format( os.path.dirname(data_item.data.relative))) - cache_dir = os.path.dirname(data_item.cache.relative) - if not os.path.exists(cache_dir): - os.makedirs(cache_dir) + if not os.path.exists(data_item.cache_dir): + os.makedirs(data_item.cache_dir) data_targets.append((input, data_item)) diff --git a/dvc/command/init.py b/dvc/command/init.py index 529c1a035..2e3568cdc 100644 --- a/dvc/command/init.py +++ b/dvc/command/init.py @@ -129,14 +129,13 @@ ProjectName = def create_empty_file(self): empty_data_path = os.path.join(self.parsed_args.data_dir, self.EMPTY_FILE_NAME) - cache_file_suffix = self.EMPTY_FILE_NAME + '_' + self.EMPTY_FILE_CHECKSUM - empty_cache_path = os.path.join(Config.CONFIG_DIR, Config.CACHE_DIR, cache_file_suffix) - open(empty_cache_path, 'w').close() - System.symlink(os.path.join('..', empty_cache_path), empty_data_path) + data_item = self.settings.path_factory.data_item(empty_data_path) + open(empty_data_path, 'w').close() + data_item.move_data_to_cache() StateFile(StateFile.COMMAND_EMPTY_FILE, - self.settings.path_factory.data_item(empty_data_path), + data_item, self.settings, input_files=[], output_files=[], diff --git a/dvc/command/run.py b/dvc/command/run.py index 662d4499c..9c8a94de7 100644 --- a/dvc/command/run.py +++ b/dvc/command/run.py @@ -112,12 +112,12 @@ class CmdRun(CmdBase): input_set = set(data_items_from_args + input_data_items) - output_set input_files_dvc = [x.data.dvc for x in input_set] - code_dependencies_dvc = self.git.abs_paths_to_dvc(code_dependencies + not_data_items_from_args) + code_dependencies_dvc = self.git.abs_paths_to_dvc(set(code_dependencies + not_data_items_from_args)) result = [] for data_item in repo_change.changed_data_items: Logger.debug('Move output file "{}" to cache dir "{}" and create a symlink'.format( - data_item.data.relative, data_item.cache.relative)) + data_item.data.relative, data_item.cache_dir)) data_item.move_data_to_cache() Logger.debug('Create state file "{}"'.format(data_item.state.relative)) diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index a88b2de10..5a84692d0 100644 --- a/dvc/data_cloud.py +++ b/dvc/data_cloud.py @@ -154,7 +154,7 @@ class DataCloudBase(object): """ pass - def _import(self, bucket, path, fname, item): + def _import(self, bucket, path, item): """ Cloud-specific method for importing data file. """ @@ -168,14 +168,13 @@ class DataCloudBase(object): """ Generic method for pulling data from the cloud """ fname = item.resolved_cache.dvc key_name = self.cache_file_key(fname) - self.create_directory(item) - return self._import(self.storage_bucket, key_name, fname, item) + return self._import(self.storage_bucket, key_name, item) def import_data(self, url, item): """ Generic method for importing data """ parsed = urlparse(url) - return self._import(parsed.netloc, parsed.path, item.cache.relative, item) + return self._import(parsed.netloc, parsed.path, item) def sync(self, fname): """ @@ -189,23 +188,6 @@ class DataCloudBase(object): else: return self.pull(item) - def create_directory(self, item): - """ Create local directories for data item cache """ - self._lock.acquire() - try: - directory = os.path.dirname(item.cache.relative) - if not os.path.exists(directory): - Logger.debug(u'Creating directory {}'.format(directory)) - try: - os.makedirs(directory) - except OSError as ex: - raise DataCloudError(u'Cannot create directory {}: {}'.format(directory, ex)) - elif not os.path.isdir(directory): - msg = u'File {} cannot be synced because {} is not a directory' - raise DataCloudError(msg.format(item.cache.relative, directory)) - finally: - self._lock.release() - def remove(self, item): """ Cloud-specific method for removing data item from the cloud. @@ -233,11 +215,11 @@ class DataCloudLOCAL(DataCloudBase): copyfile(item.resolved_cache.dvc, self.storage_path) return item - def _import(self, bucket, i, out, item): - tmp_file = self.tmp_file(out) + def _import(self, bucket, i, item): + tmp_file = self.tmp_file(item.data.dvc) try: copyfile(i, tmp_file) - os.rename(tmp_file, out) + item.import_cache(tmp_file) except Exception as exc: Logger.error('Failed to copy "{}": {}'.format(i, exc)) return None @@ -246,15 +228,15 @@ class DataCloudLOCAL(DataCloudBase): def pull(self, item): Logger.debug('sync from cloud ' + self.storage_path + " " + item.resolved_cache.dvc) - return self._import(None, self.storage_path, item.resolved_cache.dvc, item) + return self._import(None, self.storage_path, item) def remove(self, item): Logger.debug('rm from cloud ' + item.resolved_cache.dvc) os.remove(item.resolved_cache.dvc) def import_data(self, path, item): - Logger.debug('import from cloud ' + path + " " + item.cache.relative) - return self._import(None, path, item.cache.relative, item) + Logger.debug('import from cloud ' + path + " " + item.data.dvc) + return self._import(None, path, item) def _status(self, data_item): local = data_item.resolved_cache.relative @@ -366,8 +348,7 @@ class DataCloudHTTP(DataCloudBase): Download single file from url. """ - to_file = item.cache.relative - tmp_file = self.tmp_file(to_file) + tmp_file = self.tmp_file(item.data.dvc) downloaded, header = self._downloaded_size(tmp_file) req = requests.get(url, stream=True, headers=header) @@ -382,7 +363,7 @@ class DataCloudHTTP(DataCloudBase): if not self._verify_md5(req, tmp_file): return None - os.rename(tmp_file, to_file) + item.import_cache(tmp_file) return item @@ -471,10 +452,11 @@ class DataCloudAWS(DataCloudBase): """ return fname + '.download' - def _import(self, bucket_name, key_name, fname, data_item): + def _import(self, bucket_name, key_name, data_item): bucket = self._get_bucket_aws(bucket_name) + fname = data_item.data.dvc tmp_file = self.tmp_file(fname) name = os.path.basename(fname) key = bucket.get_key(key_name) @@ -494,7 +476,7 @@ class DataCloudAWS(DataCloudBase): num_retries=10) try: key.get_contents_to_filename(tmp_file, cb=create_cb(name), res_download_handler=res_h) - os.rename(tmp_file, fname) + data_item.import_cache(tmp_file) except Exception as exc: Logger.error('Failed to download "{}": {}'.format(key_name, exc)) return None @@ -714,10 +696,11 @@ class DataCloudGCP(DataCloudBase): return False - def _import(self, bucket_name, key, fname, data_item): + def _import(self, bucket_name, key, data_item): bucket = self._get_bucket_gc(bucket_name) + fname = data_item.data.dvc name = os.path.basename(fname) tmp_file = self.tmp_file(fname) @@ -738,7 +721,7 @@ class DataCloudGCP(DataCloudBase): try: blob.download_to_filename(tmp_file) - os.rename(tmp_file, fname) + data_item.import_cache(tmp_file) except Exception as exc: Logger.error('Failed to download "{}": {}'.format(key, exc)) return None diff --git a/dvc/executor.py b/dvc/executor.py index 528d8cd8c..09dcfe38a 100644 --- a/dvc/executor.py +++ b/dvc/executor.py @@ -58,5 +58,5 @@ class Executor(object): code, out, err = Executor.exec_cmd(cmd, stdout_file=stdout_file, stderr_file=stderr_file, cwd=cwd, shell=shell) if code != 0: - raise ExecutorError('Git command error ({}):\n{}'.format(' '.join(cmd), err)) + raise ExecutorError('Git command error ({}):\n{}\n{}'.format(' '.join(cmd), out, err)) return out diff --git a/dvc/git_wrapper.py b/dvc/git_wrapper.py index 2e307159e..9cc120476 100644 --- a/dvc/git_wrapper.py +++ b/dvc/git_wrapper.py @@ -179,10 +179,13 @@ class GitWrapper(GitWrapperI): Executor.exec_cmd_only_success(['git', 'add', '--all']) out_status = Executor.exec_cmd_only_success(['git', 'status', '--porcelain']) - Executor.exec_cmd_only_success(['git', 'commit', '-m', message]) - Logger.debug('[dvc-git] Commit all changes. Success.') + changes = GitWrapper.parse_porcelain_files(out_status) + if len(changes) == 0: + return [] - return GitWrapper.parse_porcelain_files(out_status) + Executor.exec_cmd_only_success(['git', 'commit', '-m', '\'{}\''.format(message)]) + Logger.debug('[dvc-git] Commit all changes. Success.') + return changes def commit_all_changes_and_log_status(self, message): statuses = self.commit_all_changes(message) diff --git a/dvc/path/data_item.py b/dvc/path/data_item.py index 90e2fbff4..6719a75a5 100644 --- a/dvc/path/data_item.py +++ b/dvc/path/data_item.py @@ -5,6 +5,7 @@ from dvc.path.path import Path from dvc.exceptions import DvcException from dvc.system import System from dvc.utils import cached_property +from dvc.data_cloud import file_md5 class DataItemError(DvcException): @@ -81,14 +82,19 @@ class DataItem(object): state_file = os.path.join(state_dir, self.data_dvc_short + self.STATE_FILE_SUFFIX) return Path(state_file, self._git) + @cached_property + def cache_dir(self): + return os.path.join(self._git.git_dir_abs, self._config.cache_dir) + @cached_property def cache(self): - cache_dir = os.path.join(self._git.git_dir_abs, self._config.cache_dir) + cache_dir = self.cache_dir if self._cache_file: file_name = os.path.relpath(os.path.realpath(self._cache_file), cache_dir) else: - file_name = self.data_dvc_short + self.CACHE_FILE_SEP + self._git.curr_commit + from dvc.state_file import StateFile + file_name = StateFile.load(self, self._git).md5 cache_file = os.path.join(cache_dir, file_name) return Path(cache_file, self._git) @@ -125,9 +131,9 @@ class DataItem(object): return os.path.relpath(self.cache.relative, data_file_dir) def move_data_to_cache(self): - cache_dir = os.path.dirname(self.cache.relative) - if not os.path.isdir(cache_dir): - os.makedirs(cache_dir) - - shutil.move(self.data.relative, self.cache.relative) + self.import_cache(self.data.relative) System.symlink(self.symlink_file, self.data.relative) + + def import_cache(self, fname): + self._cache_file = os.path.join(self.cache_dir, file_md5(fname)[0]) + os.rename(fname, self._cache_file) diff --git a/dvc/state_file.py b/dvc/state_file.py index 0b8fadd90..0b45bc625 100644 --- a/dvc/state_file.py +++ b/dvc/state_file.py @@ -42,6 +42,7 @@ class StateFile(object): PARAM_SHELL = "Shell" PARAM_TARGET_METRICS = 'TargetMetrics' TARGET_METRICS_SINGLE_METRIC = 'SingleMetric' + PARAM_MD5 = 'Md5' def __init__(self, command, @@ -56,7 +57,8 @@ class StateFile(object): stderr=None, cwd=None, shell=False, - target_metrics={}): + target_metrics={}, + md5=None): self.data_item = data_item self.settings = settings @@ -66,6 +68,10 @@ class StateFile(object): self.code_dependencies = code_dependencies self.shell = shell + self.md5 = md5 + if not md5 and data_item: + self.md5 = os.path.basename(data_item.cache.relative) + if command not in self.ACCEPTED_COMMANDS: raise StateFileError('Args error: unknown command %s' % command) self.command = command @@ -182,7 +188,8 @@ class StateFile(object): StateFile.decode_path(json.get(StateFile.PARAM_STDERR)), StateFile.decode_path(json.get(StateFile.PARAM_CWD)), json.get(StateFile.PARAM_SHELL, False), - json.get(StateFile.PARAM_TARGET_METRICS, {})) + json.get(StateFile.PARAM_TARGET_METRICS, {}), + json.get(StateFile.PARAM_MD5, '')) @staticmethod def load(data_item, settings): @@ -213,7 +220,8 @@ class StateFile(object): self.PARAM_STDOUT: self.encode_path(self.stdout), self.PARAM_STDERR: self.encode_path(self.stderr), self.PARAM_SHELL: self.shell, - self.PARAM_TARGET_METRICS: self.target_metrics + self.PARAM_TARGET_METRICS: self.target_metrics, + self.PARAM_MD5: self.md5 } if self.locked: