From 6e037a43077845f62294505a0dbaa8e3f17bf357 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 15 Jan 2018 20:30:53 +0200 Subject: [PATCH 1/2] dvc: remove show-workflow logic As discussed with @dmpetrov, removing this functionality for the upcoming release. Signed-off-by: Ruslan Kuprieiev --- dvc/cli.py | 31 ----- dvc/command/show_workflow.py | 90 -------------- dvc/graph/__init__.py | 0 dvc/graph/commit.py | 123 ------------------ dvc/graph/workflow.py | 233 ----------------------------------- tests/test_cli.py | 20 --- tests/test_workflow.py | 158 ------------------------ 7 files changed, 655 deletions(-) delete mode 100644 dvc/command/show_workflow.py delete mode 100644 dvc/graph/__init__.py delete mode 100644 dvc/graph/commit.py delete mode 100644 dvc/graph/workflow.py delete mode 100644 tests/test_workflow.py diff --git a/dvc/cli.py b/dvc/cli.py index 3b34fdf8e..065e66ec1 100644 --- a/dvc/cli.py +++ b/dvc/cli.py @@ -13,7 +13,6 @@ from dvc.command.repro import CmdRepro from dvc.command.data_sync import CmdDataPush, CmdDataPull, CmdDataStatus from dvc.command.gc import CmdGC from dvc.command.add import CmdAdd -from dvc.command.show_workflow import CmdShowWorkflow from dvc.command.config import CmdConfig from dvc.command.show_pipeline import CmdShowPipeline from dvc.command.checkout import CmdCheckout @@ -223,36 +222,6 @@ def parse_args(argv=None): help='Target data directory') pipeline_parser.set_defaults(func=CmdShowPipeline) - workflow_parser = show_subparsers.add_parser( - 'workflow', - parents=[parent_parser], - help='Show workflow image. It collapses DVC repro commits if possible.') - workflow_parser.add_argument( - 'target', - nargs='?', - help='Target metric file') - workflow_parser.add_argument( - '-d', - '--dvc-commits', - action='store_true', - default=False, - help='Show DVC repro commits.') - workflow_parser.add_argument( - '-a', - '--all-commits', - action='store_true', - default=False, - help='Show all commits') - workflow_parser.add_argument( - '-m', - '--max-commits', - metavar='M', - type=int, - default=4, - help='Max commits per graph vertex. 4 by default.') - - workflow_parser.set_defaults(func=CmdShowWorkflow) - # Checkout checkout_parser = subparsers.add_parser( 'checkout', diff --git a/dvc/command/show_workflow.py b/dvc/command/show_workflow.py deleted file mode 100644 index 59d51d0f3..000000000 --- a/dvc/command/show_workflow.py +++ /dev/null @@ -1,90 +0,0 @@ -import re -import sys -from collections import defaultdict - -from dvc.logger import Logger -from dvc.command.common.base import CmdBase -from dvc.graph.workflow import Workflow -from dvc.graph.commit import Commit -from dvc.executor import Executor, ExecutorError -from dvc.utils import parse_target_metric_file - - -class CmdShowWorkflow(CmdBase): - COMMIT_LEN = 7 - LOG_SEPARATOR = '|' - LOG_FORMAT = ['%h', '%p', '%an', '%ai', '%s'] - - def get_all_commits(self, target): - # git log --all --abbrev=7 --pretty=format:"%h|%p|%an|%ai|%s" - try: - merges_map = self.get_merges_map() - - format_str = self.LOG_SEPARATOR.join(self.LOG_FORMAT) - git_cmd = ['git', 'log', '--all', '--abbrev={}'.format(self.COMMIT_LEN), - '--pretty=format:{}'.format(format_str)] - lines = Executor.exec_cmd_only_success(git_cmd).split('\n') - - branches_multimap = self.branches_multimap() - - wf = Workflow(target, merges_map, branches_multimap) - for line in lines: - items = line.split(self.LOG_SEPARATOR, len(self.LOG_FORMAT)) - assert len(items) == 5, 'Git wrapper: git log format has {} items, 5 expected'.format(len(items)) - hash, parent_hash, name, date, comment = items - - commit = Commit(hash, parent_hash, name, date, comment, - *self.was_target_changed(hash, target), - branch_tips=branches_multimap.get(hash)) - wf.add_commit(commit) - - return wf - except ExecutorError: - raise - - def was_target_changed(self, hash, target): - git_cmd = ['git', 'show', '--pretty=', '--name-only', hash] - changed_files = set(Executor.exec_cmd_only_success(git_cmd).split('\n')) - - if target not in changed_files: - return False, None - - metric = parse_target_metric_file(target) - if metric is None: - return False, None - - return True, metric - - def get_merges_map(self): - # git log --merges --all --abbrev=7 - # {'a4b56f1': back_to_600_est} - git_cmd = ['git', 'log', '--all', '--merges', - '--abbrev={}'.format(self.COMMIT_LEN)] - # lines = map(str.strip, Executor.exec_cmd_only_success(git_cmd).split('\n')) - # lines.map - return {} - - def branches_multimap(self): - git_cmd = ['git', 'show-ref', '--abbrev={}'.format(self.COMMIT_LEN)] - lines = Executor.exec_cmd_only_success(git_cmd).split('\n') - func = unicode.split if sys.version_info[0] < 3 else str.split - items_full = map(func, lines) - items = map(lambda it: (it[0], re.sub(r'^refs/heads/', '', it[1])), items_full) - - result = defaultdict(list) - for (hash, branch) in items: - result[hash].append(branch) - return result - - def run(self): - target = self.args.target - if not target: - target = self.project.config._config['Global'].get('Target', '') - self.project.logger.debug(u'Set show workflow target as {}'.format(target)) - - wf = self.get_all_commits(target) - wf.build_graph(self.args.dvc_commits, - self.args.all_commits, - self.args.max_commits) - return 0 - diff --git a/dvc/graph/__init__.py b/dvc/graph/__init__.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/dvc/graph/commit.py b/dvc/graph/commit.py deleted file mode 100644 index 24a5a8ca2..000000000 --- a/dvc/graph/commit.py +++ /dev/null @@ -1,123 +0,0 @@ -class Commit(object): - TEXT_LIMIT = 30 - DVC_REPRO_PREFIX = 'DVC repro' - COLLAPSED_TEXT = DVC_REPRO_PREFIX + ' <>' - - def __init__(self, hash, parents, name, date, comment, - is_target=False, - target_metric=None, - branch_tips=None): - self._hash = hash - self._parent_hashes = set(parents.split()) - self._name = name - self._date = date - self._comment = comment - self._is_target = is_target - self._target_metric = target_metric - self._target_metric_delta = None - self._branch_tips = [] if branch_tips is None else branch_tips - self._collapsed_commits = [] - - self._is_collapsed = False - - @property - def hash(self): - return self._hash - - @property - def parent_hashes(self): - return self._parent_hashes - - def add_parents(self, parent_hashes): - self._parent_hashes |= parent_hashes - - def remove_parent(self, hash): - self._parent_hashes.remove(hash) - - def add_collapsed_commit(self, commit): - self._collapsed_commits.append(commit) - self._collapsed_commits += commit._collapsed_commits - - def text(self, max_commits=100): - branch_text = '' - if self._branch_tips: - branch_text = 'BRANCH TIPS: {}\n'.format(', '.join(self._branch_tips)) - - return self._text_metrics_line() + branch_text + self._comments_text(max_commits) - - @staticmethod - def _text_comment(commit): - if commit._is_collapsed: - text = Commit.COLLAPSED_TEXT - else: - text = Commit._limit_text(commit._comment) - return '[{}] {}'.format(commit.hash, text) - - @staticmethod - def _limit_text(comment): - if len(comment) < Commit.TEXT_LIMIT: - return comment - else: - return comment[:Commit.TEXT_LIMIT-3] + '...' - - def _comments_text(self, max_commits=100): - res = [self._text_comment(self)] - - for commit in self._collapsed_commits: - if not commit.is_repro: - text = self._text_comment(commit) - res.append(text) - - was_collapsed = len(res) > max_commits - res = res[:max_commits] - - max_len = max(map(lambda x: len(x), res + [""])) - res_extended_len = map(lambda x: x.ljust(max_len), res) - - if was_collapsed: - res_extended_len.append('<>') - - return '\n'.join(res_extended_len) - - def _text_metrics_line(self): - result = '' - if self._is_target and self._target_metric: - result = 'TARGET: {}'.format(self._target_metric) - if self._target_metric_delta is not None: - result += ' ({:+f})'.format(self._target_metric_delta) - result += '\n' - return result - - @property - def is_repro(self): - return self._comment.startswith(self.DVC_REPRO_PREFIX) - - def make_collapsed(self): - self._is_collapsed = True - - @property - def has_target_metric(self): - return self._is_target and self._target_metric is not None - - @property - def target_metric(self): - return self._target_metric - - def set_target_metric(self, value): - self._target_metric = value - self._is_target = True - - @property - def branch_tips(self): - return self._branch_tips - - def add_branch_tips(self, tips): - for tip in tips: - self._branch_tips.append(tip) - - def set_target_metric_delta(self, value): - self._target_metric_delta = value - - @property - def target_metric_delta(self): - return self._target_metric_delta diff --git a/dvc/graph/workflow.py b/dvc/graph/workflow.py deleted file mode 100644 index 192e76274..000000000 --- a/dvc/graph/workflow.py +++ /dev/null @@ -1,233 +0,0 @@ -import networkx as nx - -from dvc.exceptions import DvcException -from dvc.logger import Logger - - -class WorkflowError(DvcException): - def __init__(self, msg): - super(WorkflowError, self).__init__('Workflow error: ' + msg) - - -class CommitCollapseStrategy(object): - def __init__(self, workflow): - self._workflow = workflow - pass - - def is_change_needed(self, commit, child_commits): - raise NotImplementedError() - - def is_remove_needed(self, commit): - raise NotImplementedError() - - def make_collapsed(self, commit): - commit.make_collapsed() - - def upstream(self, commit): - for child in self._workflow.child_commits(commit.hash): - self.upstream_to_child(commit, child) - pass - - def upstream_to_child(self, commit, child): - if commit.has_target_metric: - if not child.has_target_metric: - child.set_target_metric(commit.target_metric) - if commit.branch_tips: - child.add_branch_tips(commit.branch_tips) - - -class CollapseDvcReproCommitsStrategy(CommitCollapseStrategy): - def __init__(self, workflow): - super(CollapseDvcReproCommitsStrategy, self).__init__(workflow) - - def is_change_needed(self, commit, hash): - return commit.is_repro - - def is_remove_needed(self, commit): - child_commits = self._workflow.child_commits(commit.hash) - return child_commits and all(ch.is_repro for ch in child_commits) - - -class CollapseNotMeticsCommitsStrategy(CommitCollapseStrategy): - def __init__(self, workflow): - super(CollapseNotMeticsCommitsStrategy, self).__init__(workflow) - - def is_change_needed(self, commit, hash): - return not commit.has_target_metric - - def is_remove_needed(self, commit): - child_commits = self._workflow.child_commits(commit.hash) - return len(child_commits) == 1 and not commit.has_target_metric - - def upstream_to_child(self, commit, child): - super(CollapseNotMeticsCommitsStrategy, self).upstream_to_child(commit, child) - if not commit.is_repro: - child.add_collapsed_commit(commit) - - def make_collapsed(self, commit): - pass - - -class Workflow(object): - def __init__(self, target, merges_map, branches_map=None): - self._target = target - self._merges_map = merges_map - self._branches_map = branches_map - - self._commits = {} - self._root_hash = None - - self._edges = {} - self._back_edges = {} - pass - - def add_commit(self, commit): - self._commits[commit.hash] = commit - - for p in commit.parent_hashes: - if p not in self._edges: - self._edges[p] = set() - self._edges[p].add(commit.hash) - if commit.hash not in self._back_edges: - self._back_edges[commit.hash] = set() - self._back_edges[commit.hash].add(p) - - if not commit.parent_hashes: - self._root_hash = commit.hash - pass - - def get_commit(self, hash): - return self._commits.get(hash) - - def child_commits(self, hash): - if hash not in self._edges: - return [] - return [self._commits[h] for h in self._edges[hash]] - - def build_graph(self, show_dvc_commits, show_all_commits, max_commits): - self.modify_workflow(show_all_commits, show_dvc_commits) - - g = nx.DiGraph(name='DVC Workflow', directed=False) - - for hash in set(list(self._edges.keys()) + list(self._back_edges.keys())): - commit = self._commits[hash] - - g.add_node(hash, - attr_dict={ - 'label': commit.text(max_commits), - 'color': self.node_color(commit) - } - ) - - for commit in self._commits.values(): - for p in commit.parent_hashes: - g.add_edge(commit.hash, p) - - A = nx.nx_agraph.to_agraph(g) - fname = 'workflow' - A = A.to_undirected() - A.write(fname + '.dot') - A.draw(fname + '.jpeg', format='jpeg', prog='dot') - pass - - def modify_workflow(self, show_all_commits, show_dvc_commits): - self.derive_target_metric_deltas() - if not show_all_commits: - self.collapse_commits(CollapseDvcReproCommitsStrategy(self)) - if not show_dvc_commits: - self.collapse_commits(CollapseNotMeticsCommitsStrategy(self)) - - @staticmethod - def node_color(commit): - if commit.target_metric_delta is not None: - if commit.target_metric_delta >= 0: - return 'green' - if commit.target_metric_delta < 0: - return 'red' - return 'black' - - def _build_graph(self, curr, g): - if '' not in self._commits or curr.hash not in self._commits: - return - next = self._commits[curr.hash] - - g.add_node(next.hash) - g.add_edge(curr.hash, next.hash) - - self._build_graph(next, g) - - def collapse_commits(self, strategy): - hashes_to_remove = [] - for commit in self._commits.values(): - if strategy.is_change_needed(commit, commit.hash): - if self._remove_or_collapse(commit, strategy): - hashes_to_remove.append(commit.hash) - - self.remove_hashes(hashes_to_remove) - - def remove_hashes(self, hashes_to_remove): - for hash in hashes_to_remove: - del self._commits[hash] - if hash in self._edges: - del self._edges[hash] - if hash in self._back_edges: - del self._back_edges[hash] - pass - - def _remove_or_collapse(self, commit, strategy): - parent_commit_hashes = self._back_edges.get(commit.hash) - child_commit_hashes = self._edges.get(commit.hash) - - if child_commit_hashes is None: - return False - - if strategy.is_remove_needed(commit): - strategy.upstream(commit) - - for hash in child_commit_hashes: - self._commits[hash].add_parents(commit.parent_hashes) - self._commits[hash].remove_parent(commit.hash) - - self._redirect_edges(self._edges, child_commit_hashes, commit.hash, parent_commit_hashes) - self._redirect_edges(self._back_edges, parent_commit_hashes, commit.hash, child_commit_hashes) - - return True - else: - strategy.make_collapsed(commit) - return False - pass - - @staticmethod - def _redirect_edges(edges, child_commit_hashes, hash, commit_hashes): - if commit_hashes: - for h in commit_hashes: - edges[h].remove(hash) - if child_commit_hashes: - edges[h] |= child_commit_hashes - - def derive_target_metric_deltas(self): - if not self._target: - return - - if self._root_hash not in self._commits: - Logger.warn('Cannot derive target metrics deltas: root commit was not found') - return - - current_target_metric = None - commit = self._commits[self._root_hash] - self._traverse_target_metric(commit, current_target_metric) - pass - - def _traverse_target_metric(self, commit, current_target_metric): - new_target_metric = current_target_metric - if commit.target_metric is not None: - new_target_metric = commit.target_metric - if current_target_metric is None: - commit.set_target_metric_delta(0.0) - else: - commit.set_target_metric_delta(new_target_metric - current_target_metric) - - if commit.hash in self._edges: - for hash in self._edges[commit.hash]: - self._traverse_target_metric(self._commits[hash], new_target_metric) - pass diff --git a/tests/test_cli.py b/tests/test_cli.py index b65affd93..8f9a8cf52 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -10,7 +10,6 @@ from dvc.command.remove import CmdRemove from dvc.command.gc import CmdGC from dvc.command.config import CmdConfig from dvc.command.show_pipeline import CmdShowPipeline -from dvc.command.show_workflow import CmdShowWorkflow from dvc.command.checkout import CmdCheckout from dvc.command.fsck import CmdFsck @@ -158,25 +157,6 @@ class TestShowPipeline(TestDvc): self.assertEqual(args.target, [target1, target2]) -class TestShowWorkflow(TestDvc): - def test(self): - target = '1' - - args = parse_args(['show', - 'workflow', - '-d', '--dvc-commits', - '-a', '--all-commits', - '-m', '1', - '--max-commits', '1', - target]) - - self.assertIsInstance(args.func(args), CmdShowWorkflow) - self.assertEqual(args.target, target) - self.assertTrue(args.dvc_commits) - self.assertTrue(args.all_commits) - self.assertEqual(args.max_commits, 1) - - class TestCheckout(TestDvc): def test(self): args = parse_args(['checkout']) diff --git a/tests/test_workflow.py b/tests/test_workflow.py deleted file mode 100644 index 0bdc9ab7a..000000000 --- a/tests/test_workflow.py +++ /dev/null @@ -1,158 +0,0 @@ -from dvc.main import main -from dvc.graph.workflow import Workflow,\ - CollapseDvcReproCommitsStrategy, CollapseNotMeticsCommitsStrategy -from dvc.graph.commit import Commit - -from tests.basic_env import TestDvc - - -class TestWorkflow(TestDvc): - def setUp(self): - super(TestWorkflow, self).setUp() - - self._commit4 = Commit('4', '3', 'name1', 'today', 'comment4') - self._commit3 = Commit('3', '2', 'name1', 'today', 'DVC repro-run ...') - self._commit2 = Commit('2', '1', 'name1', 'today', 'DVC repro-run ...') - self._commit1 = Commit('1', '', 'name1', 'today', 'comment1') - - self._showed_text1 = '[' + self._commit1.hash + '] ' + self._commit1._comment - self._showed_text2 = '[' + self._commit2.hash + '] ' + self._commit2._comment - self._showed_text3 = '[' + self._commit3.hash + '] ' + self._commit3._comment - self._showed_text4 = '[' + self._commit4.hash + '] ' + self._commit4._comment - - self._showed_text3_collapsed = '[' + self._commit3.hash + '] ' + Commit.COLLAPSED_TEXT - - def commits_basic_test(self): - self.assertFalse(self._commit1.is_repro) - self.assertTrue(self._commit2.is_repro) - self.assertTrue(self._commit3.is_repro) - self.assertFalse(self._commit4.is_repro) - pass - - def workflow_basic_test(self): - wf = Workflow('', '') - wf.add_commit(self._commit1) - wf.add_commit(self._commit2) - wf.add_commit(self._commit3) - wf.add_commit(self._commit4) - - self.assertEqual(len(wf._commits), 4) - - self.assertEqual(wf._commits['1'].text(), self._showed_text1) - self.assertEqual(wf._commits['2'].text(), self._showed_text2) - self.assertEqual(wf._commits['3'].text(), self._showed_text3) - self.assertEqual(wf._commits['4'].text(), self._showed_text4) - pass - - def collapse_test(self): - wf = Workflow('', '') - wf.add_commit(self._commit1) - wf.add_commit(self._commit2) - wf.add_commit(self._commit3) - wf.add_commit(self._commit4) - - wf.collapse_commits(CollapseDvcReproCommitsStrategy(wf)) - - self.assertEqual(len(wf._commits), 3) - self.assertEqual(wf._commits[self._commit1.hash].text(), self._showed_text1) - self.assertEqual(wf._commits[self._commit3.hash].text(), self._showed_text3_collapsed) - self.assertTrue('2' not in wf._commits) - - self.assertFalse('2' in wf._edges) - self.assertFalse('2' in wf._back_edges) - pass - - def collapse_at_dead_end_test(self): - wf = Workflow('', '') - wf.add_commit(self._commit1) - wf.add_commit(self._commit2) - wf.add_commit(self._commit3) # Dead end which cannot be collapsed - - self.assertEqual(len(wf._commits), 3) - wf.collapse_commits(CollapseDvcReproCommitsStrategy(wf)) - self.assertEqual(len(wf._commits), 2) - - self.assertTrue('1' in wf._commits) - self.assertTrue('3' in wf._commits) - self.assertFalse('2' in wf._commits) - - self.assertEqual(wf._commits[self._commit1.hash].text(), self._showed_text1) - self.assertEqual(wf._commits[self._commit3.hash].text(), self._showed_text3) - self.assertTrue('2' not in wf._commits) - pass - - def collapse_metric_commit_test(self): - value = 0.812345 - branches = ['master', 'try_smth'] - metric_commit2 = Commit('2', '1', 'name1', 'today', 'DVC repro-run ...', - True, value, branch_tips=branches) - - wf = Workflow('', '') - wf.add_commit(self._commit1) - wf.add_commit(metric_commit2) - wf.add_commit(self._commit3) - - self.assertEqual(len(wf._commits), 3) - wf.collapse_commits(CollapseDvcReproCommitsStrategy(wf)) - self.assertEqual(len(wf._commits), 2) - - self.assertEqual(wf._commits['3']._target_metric, value) - self.assertEqual(wf._commits['3'].branch_tips, branches) - pass - - def collapse_all_commits_test(self): - value = 0.812345 - branches = ['master', 'try_smth'] - not_metric_commit2 = Commit('2', '1', 'name1', 'today', 'change smth 2...', - True, branch_tips=branches) - metric_commit3 = Commit('3', '2', 'name3', 'today3', 'hello3', - True, value) - - wf = Workflow('', '') - wf.add_commit(self._commit1) - wf.add_commit(not_metric_commit2) - wf.add_commit(metric_commit3) - wf.add_commit(self._commit4) - - self.assertEqual(len(wf._commits), 4) - wf.collapse_commits(CollapseDvcReproCommitsStrategy(wf)) - self.assertEqual(len(wf._commits), 4) - wf.collapse_commits(CollapseNotMeticsCommitsStrategy(wf)) - self.assertEqual(len(wf._commits), 2) - - self.assertEqual(wf._commits['3']._target_metric, value) - self.assertEqual(wf._commits['3'].branch_tips, branches) - self.assertTrue('4' in wf._commits) - pass - - def collapsed_all_commits_comments_test(self): - value = 0.812345 - branches = ['master', 'try_smth'] - not_metric_commit2 = Commit('2', '1', 'name1', 'today', 'change smth 2...', - True, branch_tips=branches) - metric_commit3 = Commit('3', '2', 'name3', 'today3', 'hello3', - True, value) - - wf = Workflow('', '') - wf.add_commit(self._commit1) - wf.add_commit(not_metric_commit2) - wf.add_commit(metric_commit3) - wf.add_commit(self._commit4) - - wf.collapse_commits(CollapseDvcReproCommitsStrategy(wf)) - wf.collapse_commits(CollapseNotMeticsCommitsStrategy(wf)) - self.assertEqual(len(wf._commits), 2) - - def is_collapsed_commit_presented(self, commit, collapsed_commit): - self.assertTrue('[' + collapsed_commit.hash + ']' in commit.text()) - self.assertTrue(collapsed_commit._comment in commit.text()) - - is_collapsed_commit_presented(self, wf._commits['3'], metric_commit3) - is_collapsed_commit_presented(self, wf._commits['3'], not_metric_commit2) - is_collapsed_commit_presented(self, wf._commits['3'], self._commit1) - pass - - def test_cli(self): - ret = main(['show', - 'workflow']) - self.assertEqual(ret, 0) From a130a25e39994b9e0d838a20c851a94770a8d9b1 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 15 Jan 2018 21:11:25 +0200 Subject: [PATCH 2/2] dvc: refactor data_cloud Signed-off-by: Ruslan Kuprieiev --- dvc/cloud/aws.py | 326 +++++++++++++++ dvc/cloud/base.py | 118 ++++++ dvc/cloud/data_cloud.py | 114 ++++++ dvc/cloud/gcp.py | 127 ++++++ dvc/cloud/http.py | 126 ++++++ dvc/cloud/local.py | 53 +++ dvc/command/data_sync.py | 2 +- dvc/command/fsck.py | 2 +- dvc/data_cloud.py | 848 --------------------------------------- dvc/project.py | 2 +- dvc/stage.py | 2 +- dvc/utils.py | 13 + tests/test_add.py | 2 +- tests/test_config.py | 2 +- tests/test_data_cloud.py | 4 +- tests/test_gc.py | 2 +- tests/test_repro.py | 2 +- tests/test_run.py | 2 +- 18 files changed, 888 insertions(+), 859 deletions(-) create mode 100644 dvc/cloud/aws.py create mode 100644 dvc/cloud/base.py create mode 100644 dvc/cloud/data_cloud.py create mode 100644 dvc/cloud/gcp.py create mode 100644 dvc/cloud/http.py create mode 100644 dvc/cloud/local.py delete mode 100644 dvc/data_cloud.py diff --git a/dvc/cloud/aws.py b/dvc/cloud/aws.py new file mode 100644 index 000000000..d8152b82d --- /dev/null +++ b/dvc/cloud/aws.py @@ -0,0 +1,326 @@ +import os +import math + +from boto.s3.connection import S3Connection +try: + import httplib +except ImportError: + # Python3 workaround for ResumableDownloadHandler. + # See https://github.com/boto/boto/pull/3755. + import sys + import http.client as httplib + sys.modules['httplib'] = httplib +from boto.s3.resumable_download_handler import ResumableDownloadHandler + +from dvc.logger import Logger +from dvc.progress import progress +from dvc.cloud.credentials_aws import AWSCredentials +from dvc.cloud.base import DataCloudError, DataCloudBase +from dvc.utils import file_md5 + + +def sizeof_fmt(num, suffix='B'): + """ Convert number of bytes to human-readable string """ + for unit in ['', 'K', 'M', 'G', 'T', 'P', 'E', 'Z']: + if abs(num) < 1024.0: + return "%3.1f%s%s" % (num, unit, suffix) + num /= 1024.0 + return "%.1f%s%s" % (num, 'Y', suffix) + + +def percent_cb(name, part_complete, part_total, offset=0, multipart_total=None): + """ Callback for updating target progress """ + complete = offset + part_complete + total = multipart_total if multipart_total != None else part_total + + Logger.debug('{}: {} transferred out of {}'.format(name, + sizeof_fmt(complete), + sizeof_fmt(total))) + progress.update_target(os.path.basename(name), complete, total) + + +def create_cb(name, offset=0, multipart_total=None): + """ Create callback function for multipart object """ + return (lambda cur, tot: percent_cb(name, cur, tot, offset, multipart_total)) + + +class DataCloudAWS(DataCloudBase): + """ DataCloud class for Amazon Web Services """ + def __init__(self, cloud_settings): + super(DataCloudAWS, self).__init__(cloud_settings) + self._aws_creds = AWSCredentials(cloud_settings.cloud_config) + + @property + def aws_region_host(self): + """ get the region host needed for s3 access + + See notes http://docs.aws.amazon.com/general/latest/gr/rande.html#s3_region + """ + + region = self._cloud_settings.cloud_config['Region'] + if region is None or region == '': + return 's3.amazonaws.com' + if region == 'us-east-1': + return 's3.amazonaws.com' + return 's3.%s.amazonaws.com' % region + + def credential_paths(self, default): + """ + Try obtaining path to aws credentials from config file. + """ + paths = [] + credpath = self._cloud_settings.cloud_config.get('CredentialPath', None) + if credpath is not None and len(credpath) > 0: + credpath = os.path.expanduser(credpath) + if os.path.isfile(credpath): + paths.append(credpath) + else: + Logger.warn('AWS CredentialPath "%s" not found;' + 'falling back to default "%s"' % (credpath, default)) + paths.append(default) + else: + paths.append(default) + return paths + + def _get_bucket_aws(self, bucket_name): + """ get a bucket object, aws """ + if all([self._aws_creds.access_key_id, + self._aws_creds.secret_access_key, + self.aws_region_host]): + conn = S3Connection(self._aws_creds.access_key_id, + self._aws_creds.secret_access_key, + host=self.aws_region_host) + else: + conn = S3Connection() + bucket = conn.lookup(bucket_name) + if bucket is None: + raise DataCloudError('Storage path {} is not setup correctly'.format(bucket_name)) + return bucket + + @staticmethod + def _cmp_checksum(key, fname): + """ + Verify local and remote checksums. Used 'dvc-md5' metadata if supported + or falls back to etag. + """ + md5_cloud = key.metadata.get('dvc-md5', None) + md5_local = file_md5(fname)[0] + + if md5_cloud == None: + md5_cloud = key.etag[1:-1] + + if md5_cloud == md5_local: + return True + + return False + + @staticmethod + def _upload_tracker(fname): + """ + File name for upload tracker. + """ + return fname + '.upload' + + @staticmethod + def _download_tracker(fname): + """ + File name for download tracker. + """ + return fname + '.download' + + def _import(self, bucket_name, key_name, fname): + + bucket = self._get_bucket_aws(bucket_name) + + tmp_file = self.tmp_file(fname) + name = os.path.basename(fname) + key = bucket.get_key(key_name) + if not key: + Logger.error('File "{}" does not exist in the cloud'.format(key_name)) + return None + + if self._cmp_checksum(key, fname): + Logger.debug('File "{}" matches with "{}".'.format(fname, key_name)) + return fname + + Logger.debug('Downloading cache file from S3 "{}/{}" to "{}"'.format(bucket.name, + key_name, + fname)) + + res_h = ResumableDownloadHandler(tracker_file_name=self._download_tracker(tmp_file), + num_retries=10) + try: + key.get_contents_to_filename(tmp_file, cb=create_cb(name), res_download_handler=res_h) + except Exception as exc: + Logger.error('Failed to download "{}": {}'.format(key_name, exc)) + return None + + os.rename(tmp_file, fname) + + progress.finish_target(name) + Logger.debug('Downloading completed') + + return fname + + def _write_upload_tracker(self, fname, mp_id): + """ + Write multipart id to upload tracker. + """ + try: + open(self._upload_tracker(fname), 'w+').write(mp_id) + except Exception as exc: + Logger.debug("Failed to write upload tracker file for {}: {}".format(fname, exc)) + + def _unlink_upload_tracker(self, fname): + """ + Remove upload tracker file. + """ + try: + os.unlink(self._upload_tracker(fname)) + except Exception as exc: + Logger.debug("Failed to unlink upload tracker file for {}: {}".format(fname, exc)) + + def _resume_multipart(self, key, fname): + """ + Try resuming multipart upload. + """ + try: + mp_id = open(self._upload_tracker(fname), 'r').read() + except Exception as exc: + Logger.debug("Failed to read upload tracker file for {}: {}".format(fname, exc)) + return None + + for part in key.bucket.get_all_multipart_uploads(): + if part.id != mp_id: + continue + + Logger.debug("Found existing multipart {}".format(mp_id)) + return part + + return None + + def _create_multipart(self, key, fname): + """ + Create multipart upload and save info to tracker file. + """ + # AWS doesn't provide easilly accessible md5 for multipart + # objects, so we have to store our own md5 sum to use later. + metadata = {'dvc-md5' : str(file_md5(fname)[0])} + multipart = key.bucket.initiate_multipart_upload(key.name, metadata=metadata) + self._write_upload_tracker(fname, multipart.id) + return multipart + + def _get_multipart(self, key, fname): + """ + Try resuming multipart upload if supported. + """ + multipart = self._resume_multipart(key, fname) + if multipart != None: + return multipart + + return self._create_multipart(key, fname) + + @staticmethod + def _skip_part(multipart, part_num, size): + """ + Skip part of multipart upload if it has been already uploaded to the server. + """ + for part in multipart.get_all_parts(): + if part.part_number == part_num and part.size == size:# and p.etag and p.last_modified + Logger.debug("Skipping part #{}".format(str(part_num))) + return True + return False + + def _push_multipart(self, key, fname): + """ + Upload local file to cloud as a multipart upload. + """ + multipart = self._get_multipart(key, fname) + + source_size = os.stat(fname).st_size + chunk_size = 50*1024*1024 + chunk_count = int(math.ceil(source_size / float(chunk_size))) + + with open(fname, 'rb') as fobj: + for i in range(chunk_count): + offset = i * chunk_size + left = source_size - offset + size = min([chunk_size, left]) + part_num = i + 1 + + if self._skip_part(multipart, part_num, size): + continue + + fobj.seek(offset) + multipart.upload_part_from_file(fp=fobj, + replace=False, + size=size, + num_cb=100, + part_num=part_num, + cb=create_cb(fname, offset, source_size)) + + if len(multipart.get_all_parts()) != chunk_count: + raise Exception("Couldn't upload all file parts") + + multipart.complete_upload() + self._unlink_upload_tracker(fname) + + def push(self, path): + """ push, aws version """ + + aws_key = self.cache_file_key(path) + bucket = self._get_bucket_aws(self.storage_bucket) + key = bucket.get_key(aws_key) + if key: + Logger.debug('File already uploaded to the cloud. Checksum validation...') + + if self._cmp_checksum(key, path): + Logger.debug('File checksum matches. No uploading is needed.') + return path + + Logger.debug('Checksum miss-match. Re-uploading is required.') + + key = bucket.new_key(aws_key) + + try: + self._push_multipart(key, path) + except Exception as exc: + Logger.error('Failed to upload "{}": {}'.format(path, exc)) + return None + + progress.finish_target(os.path.basename(path)) + + return path + + def _status(self, path): + aws_key = self.cache_file_key(path) + bucket = self._get_bucket_aws(self.storage_bucket) + key = bucket.get_key(aws_key) + + remote_exists = key is not None + local_exists = os.path.exists(path) + diff = None + if remote_exists and local_exists: + diff = self._cmp_checksum(key, path) + + return (local_exists, remote_exists, diff) + + def remove(self, path): + aws_file_name = self.cache_file_key(path) + + Logger.debug(u'[Cmd-Remove] Remove from cloud {}.'.format(aws_file_name)) + + if not self._aws_creds.access_key_id or not self._aws_creds.secret_access_key: + Logger.debug('[Cmd-Remove] Unable to check cache file in the cloud') + return + conn = S3Connection(self._aws_creds.access_key_id, self._aws_creds.secret_access_key) + bucket_name = self.storage_bucket + bucket = conn.lookup(bucket_name) + if bucket: + key = bucket.get_key(aws_file_name) + if not key: + Logger.warn('[Cmd-Remove] S3 remove warning: ' + 'file "{}" does not exist in S3'.format(aws_file_name)) + else: + key.delete() + Logger.info('[Cmd-Remove] File "{}" was removed from S3'.format(aws_file_name)) diff --git a/dvc/cloud/base.py b/dvc/cloud/base.py new file mode 100644 index 000000000..fa267578e --- /dev/null +++ b/dvc/cloud/base.py @@ -0,0 +1,118 @@ +import os + +from dvc.logger import Logger +from dvc.exceptions import DvcException +from dvc.config import ConfigError + + +STATUS_UNKNOWN = 0 +STATUS_OK = 1 +STATUS_MODIFIED = 2 +STATUS_NEW = 3 +STATUS_DELETED = 4 + + +STATUS_MAP = { + # (local_exists, remote_exists, cmp) + (True, True, True) : STATUS_OK, + (True, True, False) : STATUS_MODIFIED, + (True, False, None) : STATUS_NEW, + (False, True, None) : STATUS_DELETED, +} + + +class DataCloudError(DvcException): + """ Data Cloud exception """ + def __init__(self, msg): + super(DataCloudError, self).__init__('Data sync error: {}'.format(msg)) + + +class DataCloudBase(object): + """ Base class for DataCloud """ + def __init__(self, cloud_settings): + self._cloud_settings = cloud_settings + + @property + def storage_path(self): + """ get storage path + + Precedence: Storage, then cloud specific + """ + + if self._cloud_settings.global_storage_path: + return self._cloud_settings.global_storage_path + + path = self._cloud_settings.cloud_config.get('StoragePath', None) + if path is None: + raise ConfigError('invalid StoragePath: not set for Data or cloud specific') + + return path + + def _storage_path_parts(self): + """ + Split storage path into parts. I.e. 'dvc-test/myrepo' -> ['dvc', 'myrepo'] + """ + return self.storage_path.strip('/').split('/', 1) + + @property + def storage_bucket(self): + """ Data -> StoragePath takes precedence; if doesn't exist, use cloud-specific """ + return self._storage_path_parts()[0] + + @property + def storage_prefix(self): + """ + Prefix within the bucket. I.e. 'myrepo' in 'dvc-test/myrepo'. + """ + parts = self._storage_path_parts() + if len(parts) > 1: + return parts[1] + return '' + + def cache_file_key(self, fname): + """ Key of a file within the bucket """ + return '{}/{}'.format(self.storage_prefix, os.path.basename(fname)).strip('/') + + @staticmethod + def tmp_file(fname): + """ Temporary name for a partial download """ + return fname + '.part' + + def sanity_check(self): + """ + Cloud-specific method to check config for basic requirements. + """ + pass + + def _import(self, bucket, fin, fout): + """ + Cloud-specific method for importing data file. + """ + pass + + def push(self, path): + """ Cloud-specific method for pushing data """ + pass + + def pull(self, path): + """ Generic method for pulling data from the cloud """ + key_name = self.cache_file_key(path) + return self._import(self.storage_bucket, key_name, path) + + def remove(self, path): + """ + Cloud-specific method for removing data item from the cloud. + """ + pass + + def _status(self, path): + """ + Cloud-specific method for checking data item status. + """ + pass + + def status(self, path): + """ + Generic method for checking data item status. + """ + return STATUS_MAP.get(self._status(path), STATUS_UNKNOWN) diff --git a/dvc/cloud/data_cloud.py b/dvc/cloud/data_cloud.py new file mode 100644 index 000000000..02727e94a --- /dev/null +++ b/dvc/cloud/data_cloud.py @@ -0,0 +1,114 @@ +from dvc.cloud.instance_manager import CloudSettings +from dvc.logger import Logger +from dvc.exceptions import DvcException +from dvc.config import ConfigError +from dvc.utils import map_progress + +from dvc.cloud.aws import DataCloudAWS +from dvc.cloud.gcp import DataCloudGCP +from dvc.cloud.local import DataCloudLOCAL +from dvc.cloud.http import DataCloudHTTP + + +class DataCloud(object): + """ Generic class to do initial config parsing and redirect to proper DataCloud methods """ + + CLOUD_MAP = { + 'AWS' : DataCloudAWS, + 'GCP' : DataCloudGCP, + 'HTTP' : DataCloudHTTP, + 'LOCAL' : DataCloudLOCAL, + } + + SCHEME_MAP = { + 's3' : 'AWS', + 'http' : 'HTTP', + 'https' : 'HTTP', + 'ftp' : 'HTTP', + 'gs' : 'GCP', + '' : 'LOCAL', + } + + def __init__(self, config): + self._config = config + + cloud_type = self._config['Global'].get('Cloud', '').strip().upper() + if cloud_type not in self.CLOUD_MAP.keys(): + raise ConfigError('Wrong cloud type %s specified' % cloud_type) + + if cloud_type not in self._config.keys(): + raise ConfigError('Can\'t find cloud section \'[%s]\' in config' % cloud_type) + + cloud_settings = self.get_cloud_settings(self._config, + cloud_type) + + self.typ = cloud_type + self._cloud = self.CLOUD_MAP[cloud_type](cloud_settings) + + self.sanity_check() + + @staticmethod + def get_cloud_settings(config, cloud_type): + """ + Obtain cloud settings from config. + """ + if cloud_type not in config.keys(): + cloud_config = None + else: + cloud_config = config[cloud_type] + global_storage_path = config['Global'].get('StoragePath', None) + cloud_settings = CloudSettings(global_storage_path, cloud_config) + return cloud_settings + + def sanity_check(self): + """ sanity check a config + + check that we have a cloud and storagePath + if aws, check can read credentials + if google, check ProjectName + + Returns: + (T,) if good + (F, issues) if bad + """ + key = 'Cloud' + if key.lower() not in [k.lower() for k in self._config['Global'].keys()] or len(self._config['Global'][key]) < 1: + raise ConfigError('Please set %s in section Global in config file' % key) + + # now that a cloud is chosen, can check StoragePath + storage_path = self._cloud.storage_path + if storage_path is None or len(storage_path) == 0: + raise ConfigError('Please set StoragePath = bucket/{optional path} ' + 'in config file in a cloud specific section') + + self._cloud.sanity_check() + + def _map_targets(self, func, targets, jobs): + """ + Process targets as data items in parallel. + """ + return map_progress(func, targets, jobs) + + def sync(self, targets, jobs=1): + """ + Sync data items in a cloud-agnostic way. + """ + return self._map_targets(self._cloud.sync, targets, jobs) + + def push(self, targets, jobs=1): + """ + Push data items in a cloud-agnostic way. + """ + return self._map_targets(self._cloud.push, targets, jobs) + + def pull(self, targets, jobs=1): + """ + Pull data items in a cloud-agnostic way. + """ + return self._map_targets(self._cloud.pull, targets, jobs) + + def status(self, targets, jobs=1): + """ + Check status of data items in a cloud-agnostic way. + """ + return self._map_targets(self._cloud.status, targets, jobs) diff --git a/dvc/cloud/gcp.py b/dvc/cloud/gcp.py new file mode 100644 index 000000000..73c16cbf7 --- /dev/null +++ b/dvc/cloud/gcp.py @@ -0,0 +1,127 @@ +import base64 +import os + +from google.cloud import storage as gc + +from dvc.logger import Logger +from dvc.config import ConfigError +from dvc.cloud.base import DataCloudError, DataCloudBase +from dvc.utils import file_md5 +from dvc.progress import progress + + +class DataCloudGCP(DataCloudBase): + """ DataCloud class for Google Cloud Platform """ + @property + def gc_project_name(self): + """ + Get project name from config. + """ + return self._cloud_settings.cloud_config.get('ProjectName', None) + + def sanity_check(self): + project = self.gc_project_name + if project is None or len(project) < 1: + raise ConfigError('can\'t read google cloud project name. ' + 'Please set ProjectName in section GC.') + + def _get_bucket_gc(self, storage_bucket): + """ get a bucket object, gc """ + client = gc.Client(project=self.gc_project_name) + bucket = client.bucket(storage_bucket) + if not bucket.exists(): + raise DataCloudError('sync up: google cloud bucket {} ' + 'doesn\'t exist'.format(self.storage_bucket)) + return bucket + + @staticmethod + def _cmp_checksum(blob, fname): + """ + Verify local and remote checksums. + """ + md5 = file_md5(fname)[1] + b64_encoded_md5 = base64.b64encode(md5).decode() if md5 else None + + if blob.md5_hash == b64_encoded_md5: + return True + + return False + + def _import(self, bucket_name, key, fname): + + bucket = self._get_bucket_gc(bucket_name) + + name = os.path.basename(fname) + tmp_file = self.tmp_file(fname) + + blob = bucket.get_blob(key) + if not blob: + Logger.error('File "{}" does not exist in the cloud'.format(key)) + return None + + if self._cmp_checksum(blob, fname): + Logger.debug('File "{}" matches with "{}".'.format(fname, key)) + return fname + + Logger.debug('Downloading cache file from gc "{}/{}"'.format(bucket.name, key)) + + # 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) + + try: + blob.download_to_filename(tmp_file) + except Exception as exc: + Logger.error('Failed to download "{}": {}'.format(key, exc)) + return None + + os.rename(tmp_file, fname) + + progress.finish_target(name) + + Logger.debug('Downloading completed') + + return fname + + def push(self, path): + """ push, gcp version """ + + bucket = self._get_bucket_gc(self.storage_bucket) + blob_name = self.cache_file_key(path) + name = os.path.basename(path) + + blob = bucket.get_blob(blob_name) + if blob is not None and blob.exists(): + if self._cmp_checksum(blob, path): + Logger.debug('checksum %s matches. Skipping upload' % path) + return path + Logger.debug('checksum %s mismatch. re-uploading' % path) + + # same as in _import + progress.update_target(name, 0, None) + + blob = bucket.blob(blob_name) + blob.upload_from_filename(path) + + progress.finish_target(name) + Logger.debug('uploading %s completed' % path) + + return path + + def _status(self, path): + """ status, gcp version """ + + bucket = self._get_bucket_gc(self.storage_bucket) + blob_name = self.cache_file_key(path) + blob = bucket.get_blob(blob_name) + + remote_exists = blob is not None and blob.exists() + local_exists = os.path.exists(path) + diff = None + if remote_exists and local_exists: + diff = self._cmp_checksum(blob, path) + + return (local_exists, remote_exists, diff) + + def remove(self, item): + raise Exception('NOT IMPLEMENTED YET') diff --git a/dvc/cloud/http.py b/dvc/cloud/http.py new file mode 100644 index 000000000..506660944 --- /dev/null +++ b/dvc/cloud/http.py @@ -0,0 +1,126 @@ +import os +import requests + +from dvc.logger import Logger +from dvc.progress import progress +from dvc.utils import copyfile +from dvc.cloud.base import DataCloudError, DataCloudBase +from dvc.utils import file_md5 + + +class DataCloudHTTP(DataCloudBase): + """ + Driver for http cloud. + """ + def push(self, path): + raise Exception('Not implemented yet') + + def pull(self, path): + raise Exception('Not implemented yet') + + def remove(self, path): + raise Exception('Not implemented yet') + + def status(self, path): + raise Exception('Not implemented yet') + + @staticmethod + def _downloaded_size(fname): + """ + Check how much is already downloaded. + """ + if os.path.exists(fname): + downloaded = os.path.getsize(fname) + header = {'Range': 'bytes=%d-' % downloaded} + + Logger.debug('found existing {} file, resuming download'.format(fname)) + + return (downloaded, header) + + return (0, None) + + @staticmethod + def _get_header(req, name): + """ + Get header value from request. + """ + val = req.headers.get(name) + if val == None: + Logger.debug('\'{}\' not supported by the server'.format(name)) + + return val + + def _verify_downloaded_size(self, req, downloaded_size): + """ + Check that server supports resuming downloads. + """ + content_range = self._get_header(req, 'content-range') + if downloaded_size and content_range == None: + Logger.debug('Can\'t resume download') + return 0 + + return downloaded_size + + def _download(self, req, fname, downloaded): + """ + Download file with progress bar. + """ + mode = 'ab' if downloaded else 'wb' + name = os.path.basename(req.url) + total_length = self._get_header(req, 'content-length') + chunk_size = 1024 * 100 + + progress.update_target(name, downloaded, total_length) + + with open(fname, mode) as fobj: + for chunk in req.iter_content(chunk_size=chunk_size): + if not chunk: # filter out keep-alive new chunks + continue + + fobj.write(chunk) + downloaded += len(chunk) + progress.update_target(name, downloaded, total_length) + + + progress.finish_target(name) + + def _verify_md5(self, req, fname): + """ + Verify md5 of a downloaded file if server supports 'content-md5' header. + """ + md5 = file_md5(fname)[0] + content_md5 = self._get_header(req, 'content-md5') + + if content_md5 == None: + return True + + if md5 != content_md5: + Logger.error('Checksum mismatch') + return False + + Logger.debug('Checksum matches') + return True + + def import_data(self, url, path): + """ + Download single file from url. + """ + + tmp_file = self.tmp_file(path) + + downloaded, header = self._downloaded_size(tmp_file) + req = requests.get(url, stream=True, headers=header) + downloaded = self._verify_downloaded_size(req, downloaded) + + try: + self._download(req, tmp_file, downloaded) + except Exception as exc: + Logger.error('Failed to download "{}": {}'.format(url, exc)) + return None + + if not self._verify_md5(req, tmp_file): + return None + + os.rename(tmp_file, path) + + return path diff --git a/dvc/cloud/local.py b/dvc/cloud/local.py new file mode 100644 index 000000000..ba6a1d73b --- /dev/null +++ b/dvc/cloud/local.py @@ -0,0 +1,53 @@ +import os +import filecmp + +from dvc.logger import Logger +from dvc.utils import copyfile, file_md5 +from dvc.cloud.base import DataCloudError, DataCloudBase + + +class DataCloudLOCAL(DataCloudBase): + """ + Driver for local storage. + """ + def push(self, path): + Logger.debug('sync to cloud ' + path + " " + self.storage_path) + copyfile(path, self.storage_path) + return path + + def _import(self, bucket, i, path): + inp = os.path.join(self.storage_path, i) + tmp_file = self.tmp_file(path) + try: + copyfile(inp, tmp_file) + except Exception as exc: + Logger.error('Failed to copy "{}": {}'.format(i, exc)) + return None + + os.rename(tmp_file, path) + + return path + + def pull(self, path): + Logger.debug('sync from cloud ' + path) + return self._import(None, path, path) + + def remove(self, path): + Logger.debug('rm from cloud ' + path) + os.remove(path) + + def import_data(self, path, out): + Logger.debug('import from cloud ' + path + " " + out) + return self._import(None, path, out) + + def _status(self, path): + local = path + remote = '{}/{}'.format(self.storage_path, os.path.basename(local)) + + remote_exists = os.path.exists(remote) + local_exists = os.path.exists(local) + diff = None + if local_exists and remote_exists: + diff = filecmp.cmp(local, remote) + + return (local_exists, remote_exists, diff) diff --git a/dvc/command/data_sync.py b/dvc/command/data_sync.py index 6dc146de0..d79b379ab 100644 --- a/dvc/command/data_sync.py +++ b/dvc/command/data_sync.py @@ -1,4 +1,4 @@ -import dvc.data_cloud as cloud +import dvc.cloud.data_cloud as cloud from dvc.command.common.base import CmdBase class CmdDataPull(CmdBase): diff --git a/dvc/command/fsck.py b/dvc/command/fsck.py index a14614ecc..c3846d53c 100644 --- a/dvc/command/fsck.py +++ b/dvc/command/fsck.py @@ -2,7 +2,7 @@ import os from dvc import stage from dvc.command.common.base import CmdBase -from dvc.data_cloud import file_md5 +from dvc.utils import file_md5 class FsckFileDep(object): diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py deleted file mode 100644 index 7e217a4a5..000000000 --- a/dvc/data_cloud.py +++ /dev/null @@ -1,848 +0,0 @@ -""" -Data cloud(Local, AWS, GCP) drivers. -""" -import base64 -import hashlib -import os -import threading -import requests -import filecmp -import math - -from boto.s3.connection import S3Connection -try: - import httplib -except ImportError: - # Python3 workaround for ResumableDownloadHandler. - # See https://github.com/boto/boto/pull/3755. - import sys - import http.client as httplib - sys.modules['httplib'] = httplib -from boto.s3.resumable_download_handler import ResumableDownloadHandler -from google.cloud import storage as gc - -from dvc.cloud.instance_manager import CloudSettings -from dvc.logger import Logger -from dvc.exceptions import DvcException -from dvc.config import ConfigError -from dvc.progress import progress -from dvc.utils import copyfile -from dvc.cloud.credentials_aws import AWSCredentials -from dvc.system import System -from dvc.utils import map_progress - -STATUS_UNKNOWN = 0 -STATUS_OK = 1 -STATUS_MODIFIED = 2 -STATUS_NEW = 3 -STATUS_DELETED = 4 - -STATUS_MAP = { - # (local_exists, remote_exists, cmp) - (True, True, True) : STATUS_OK, - (True, True, False) : STATUS_MODIFIED, - (True, False, None) : STATUS_NEW, - (False, True, None) : STATUS_DELETED, -} - -class DataCloudError(DvcException): - """ Data Cloud exception """ - def __init__(self, msg): - super(DataCloudError, self).__init__('Data sync error: {}'.format(msg)) - - -def sizeof_fmt(num, suffix='B'): - """ Convert number of bytes to human-readable string """ - for unit in ['', 'K', 'M', 'G', 'T', 'P', 'E', 'Z']: - if abs(num) < 1024.0: - return "%3.1f%s%s" % (num, unit, suffix) - num /= 1024.0 - return "%.1f%s%s" % (num, 'Y', suffix) - - -def percent_cb(name, part_complete, part_total, offset=0, multipart_total=None): - """ Callback for updating target progress """ - complete = offset + part_complete - total = multipart_total if multipart_total != None else part_total - - Logger.debug('{}: {} transferred out of {}'.format(name, - sizeof_fmt(complete), - sizeof_fmt(total))) - progress.update_target(os.path.basename(name), complete, total) - - -def create_cb(name, offset=0, multipart_total=None): - """ Create callback function for multipart object """ - return (lambda cur, tot: percent_cb(name, cur, tot, offset, multipart_total)) - - -def file_md5(fname): - """ get the (md5 hexdigest, md5 digest) of a file """ - if os.path.exists(fname): - hash_md5 = hashlib.md5() - with open(fname, "rb") as fobj: - for chunk in iter(lambda: fobj.read(1024*1000), b""): - hash_md5.update(chunk) - return (hash_md5.hexdigest(), hash_md5.digest()) - else: - return (None, None) - - -class DataCloudBase(object): - """ Base class for DataCloud """ - def __init__(self, cloud_settings): - self._cloud_settings = cloud_settings - self._lock = threading.Lock() - - @property - def storage_path(self): - """ get storage path - - Precedence: Storage, then cloud specific - """ - - if self._cloud_settings.global_storage_path: - return self._cloud_settings.global_storage_path - - path = self._cloud_settings.cloud_config.get('StoragePath', None) - if path is None: - raise ConfigError('invalid StoragePath: not set for Data or cloud specific') - - return path - - def _storage_path_parts(self): - """ - Split storage path into parts. I.e. 'dvc-test/myrepo' -> ['dvc', 'myrepo'] - """ - return self.storage_path.strip('/').split('/', 1) - - @property - def storage_bucket(self): - """ Data -> StoragePath takes precedence; if doesn't exist, use cloud-specific """ - return self._storage_path_parts()[0] - - @property - def storage_prefix(self): - """ - Prefix within the bucket. I.e. 'myrepo' in 'dvc-test/myrepo'. - """ - parts = self._storage_path_parts() - if len(parts) > 1: - return parts[1] - return '' - - def cache_file_key(self, fname): - """ Key of a file within the bucket """ - return '{}/{}'.format(self.storage_prefix, os.path.basename(fname)).strip('/') - - @staticmethod - def tmp_file(fname): - """ Temporary name for a partial download """ - return fname + '.part' - - def sanity_check(self): - """ - Cloud-specific method to check config for basic requirements. - """ - pass - - def _import(self, bucket, fin, fout): - """ - Cloud-specific method for importing data file. - """ - pass - - def push(self, path): - """ Cloud-specific method for pushing data """ - pass - - def pull(self, path): - """ Generic method for pulling data from the cloud """ - key_name = self.cache_file_key(path) - return self._import(self.storage_bucket, key_name, path) - - def remove(self, path): - """ - Cloud-specific method for removing data item from the cloud. - """ - pass - - def _status(self, path): - """ - Cloud-specific method for checking data item status. - """ - pass - - def status(self, path): - """ - Generic method for checking data item status. - """ - return STATUS_MAP.get(self._status(path), STATUS_UNKNOWN) - - -class DataCloudLOCAL(DataCloudBase): - """ - Driver for local storage. - """ - def push(self, path): - Logger.debug('sync to cloud ' + path + " " + self.storage_path) - copyfile(path, self.storage_path) - return path - - def _import(self, bucket, i, path): - inp = os.path.join(self.storage_path, i) - tmp_file = self.tmp_file(path) - try: - copyfile(inp, tmp_file) - except Exception as exc: - Logger.error('Failed to copy "{}": {}'.format(i, exc)) - return None - - os.rename(tmp_file, path) - - return path - - def pull(self, path): - Logger.debug('sync from cloud ' + path) - return self._import(None, path, path) - - def remove(self, path): - Logger.debug('rm from cloud ' + path) - os.remove(path) - - def import_data(self, path, out): - Logger.debug('import from cloud ' + path + " " + out) - return self._import(None, path, out) - - def _status(self, path): - local = path - remote = '{}/{}'.format(self.storage_path, os.path.basename(local)) - - remote_exists = os.path.exists(remote) - local_exists = os.path.exists(local) - diff = None - if local_exists and remote_exists: - diff = filecmp.cmp(local, remote) - - return (local_exists, remote_exists, diff) - - -class DataCloudHTTP(DataCloudBase): - """ - Driver for http cloud. - """ - def push(self, path): - raise Exception('Not implemented yet') - - def pull(self, path): - raise Exception('Not implemented yet') - - def remove(self, path): - raise Exception('Not implemented yet') - - def status(self, path): - raise Exception('Not implemented yet') - - @staticmethod - def _downloaded_size(fname): - """ - Check how much is already downloaded. - """ - if os.path.exists(fname): - downloaded = os.path.getsize(fname) - header = {'Range': 'bytes=%d-' % downloaded} - - Logger.debug('found existing {} file, resuming download'.format(fname)) - - return (downloaded, header) - - return (0, None) - - @staticmethod - def _get_header(req, name): - """ - Get header value from request. - """ - val = req.headers.get(name) - if val == None: - Logger.debug('\'{}\' not supported by the server'.format(name)) - - return val - - def _verify_downloaded_size(self, req, downloaded_size): - """ - Check that server supports resuming downloads. - """ - content_range = self._get_header(req, 'content-range') - if downloaded_size and content_range == None: - Logger.debug('Can\'t resume download') - return 0 - - return downloaded_size - - def _download(self, req, fname, downloaded): - """ - Download file with progress bar. - """ - mode = 'ab' if downloaded else 'wb' - name = os.path.basename(req.url) - total_length = self._get_header(req, 'content-length') - chunk_size = 1024 * 100 - - progress.update_target(name, downloaded, total_length) - - with open(fname, mode) as fobj: - for chunk in req.iter_content(chunk_size=chunk_size): - if not chunk: # filter out keep-alive new chunks - continue - - fobj.write(chunk) - downloaded += len(chunk) - progress.update_target(name, downloaded, total_length) - - - progress.finish_target(name) - - def _verify_md5(self, req, fname): - """ - Verify md5 of a downloaded file if server supports 'content-md5' header. - """ - md5 = file_md5(fname)[0] - content_md5 = self._get_header(req, 'content-md5') - - if content_md5 == None: - return True - - if md5 != content_md5: - Logger.error('Checksum mismatch') - return False - - Logger.debug('Checksum matches') - return True - - def import_data(self, url, path): - """ - Download single file from url. - """ - - tmp_file = self.tmp_file(path) - - downloaded, header = self._downloaded_size(tmp_file) - req = requests.get(url, stream=True, headers=header) - downloaded = self._verify_downloaded_size(req, downloaded) - - try: - self._download(req, tmp_file, downloaded) - except Exception as exc: - Logger.error('Failed to download "{}": {}'.format(url, exc)) - return None - - if not self._verify_md5(req, tmp_file): - return None - - os.rename(tmp_file, path) - - return path - - -class DataCloudAWS(DataCloudBase): - """ DataCloud class for Amazon Web Services """ - def __init__(self, cloud_settings): - super(DataCloudAWS, self).__init__(cloud_settings) - self._aws_creds = AWSCredentials(cloud_settings.cloud_config) - - @property - def aws_region_host(self): - """ get the region host needed for s3 access - - See notes http://docs.aws.amazon.com/general/latest/gr/rande.html#s3_region - """ - - region = self._cloud_settings.cloud_config['Region'] - if region is None or region == '': - return 's3.amazonaws.com' - if region == 'us-east-1': - return 's3.amazonaws.com' - return 's3.%s.amazonaws.com' % region - - def credential_paths(self, default): - """ - Try obtaining path to aws credentials from config file. - """ - paths = [] - credpath = self._cloud_settings.cloud_config.get('CredentialPath', None) - if credpath is not None and len(credpath) > 0: - credpath = os.path.expanduser(credpath) - if os.path.isfile(credpath): - paths.append(credpath) - else: - Logger.warn('AWS CredentialPath "%s" not found;' - 'falling back to default "%s"' % (credpath, default)) - paths.append(default) - else: - paths.append(default) - return paths - - def _get_bucket_aws(self, bucket_name): - """ get a bucket object, aws """ - if all([self._aws_creds.access_key_id, - self._aws_creds.secret_access_key, - self.aws_region_host]): - conn = S3Connection(self._aws_creds.access_key_id, - self._aws_creds.secret_access_key, - host=self.aws_region_host) - else: - conn = S3Connection() - bucket = conn.lookup(bucket_name) - if bucket is None: - raise DataCloudError('Storage path {} is not setup correctly'.format(bucket_name)) - return bucket - - @staticmethod - def _cmp_checksum(key, fname): - """ - Verify local and remote checksums. Used 'dvc-md5' metadata if supported - or falls back to etag. - """ - md5_cloud = key.metadata.get('dvc-md5', None) - md5_local = file_md5(fname)[0] - - if md5_cloud == None: - md5_cloud = key.etag[1:-1] - - if md5_cloud == md5_local: - return True - - return False - - @staticmethod - def _upload_tracker(fname): - """ - File name for upload tracker. - """ - return fname + '.upload' - - @staticmethod - def _download_tracker(fname): - """ - File name for download tracker. - """ - return fname + '.download' - - def _import(self, bucket_name, key_name, fname): - - bucket = self._get_bucket_aws(bucket_name) - - tmp_file = self.tmp_file(fname) - name = os.path.basename(fname) - key = bucket.get_key(key_name) - if not key: - Logger.error('File "{}" does not exist in the cloud'.format(key_name)) - return None - - if self._cmp_checksum(key, fname): - Logger.debug('File "{}" matches with "{}".'.format(fname, key_name)) - return fname - - Logger.debug('Downloading cache file from S3 "{}/{}" to "{}"'.format(bucket.name, - key_name, - fname)) - - res_h = ResumableDownloadHandler(tracker_file_name=self._download_tracker(tmp_file), - num_retries=10) - try: - key.get_contents_to_filename(tmp_file, cb=create_cb(name), res_download_handler=res_h) - except Exception as exc: - Logger.error('Failed to download "{}": {}'.format(key_name, exc)) - return None - - os.rename(tmp_file, fname) - - progress.finish_target(name) - Logger.debug('Downloading completed') - - return fname - - def _write_upload_tracker(self, fname, mp_id): - """ - Write multipart id to upload tracker. - """ - try: - open(self._upload_tracker(fname), 'w+').write(mp_id) - except Exception as exc: - Logger.debug("Failed to write upload tracker file for {}: {}".format(fname, exc)) - - def _unlink_upload_tracker(self, fname): - """ - Remove upload tracker file. - """ - try: - os.unlink(self._upload_tracker(fname)) - except Exception as exc: - Logger.debug("Failed to unlink upload tracker file for {}: {}".format(fname, exc)) - - def _resume_multipart(self, key, fname): - """ - Try resuming multipart upload. - """ - try: - mp_id = open(self._upload_tracker(fname), 'r').read() - except Exception as exc: - Logger.debug("Failed to read upload tracker file for {}: {}".format(fname, exc)) - return None - - for part in key.bucket.get_all_multipart_uploads(): - if part.id != mp_id: - continue - - Logger.debug("Found existing multipart {}".format(mp_id)) - return part - - return None - - def _create_multipart(self, key, fname): - """ - Create multipart upload and save info to tracker file. - """ - # AWS doesn't provide easilly accessible md5 for multipart - # objects, so we have to store our own md5 sum to use later. - metadata = {'dvc-md5' : str(file_md5(fname)[0])} - multipart = key.bucket.initiate_multipart_upload(key.name, metadata=metadata) - self._write_upload_tracker(fname, multipart.id) - return multipart - - def _get_multipart(self, key, fname): - """ - Try resuming multipart upload if supported. - """ - multipart = self._resume_multipart(key, fname) - if multipart != None: - return multipart - - return self._create_multipart(key, fname) - - @staticmethod - def _skip_part(multipart, part_num, size): - """ - Skip part of multipart upload if it has been already uploaded to the server. - """ - for part in multipart.get_all_parts(): - if part.part_number == part_num and part.size == size:# and p.etag and p.last_modified - Logger.debug("Skipping part #{}".format(str(part_num))) - return True - return False - - def _push_multipart(self, key, fname): - """ - Upload local file to cloud as a multipart upload. - """ - multipart = self._get_multipart(key, fname) - - source_size = os.stat(fname).st_size - chunk_size = 50*1024*1024 - chunk_count = int(math.ceil(source_size / float(chunk_size))) - - with open(fname, 'rb') as fobj: - for i in range(chunk_count): - offset = i * chunk_size - left = source_size - offset - size = min([chunk_size, left]) - part_num = i + 1 - - if self._skip_part(multipart, part_num, size): - continue - - fobj.seek(offset) - multipart.upload_part_from_file(fp=fobj, - replace=False, - size=size, - num_cb=100, - part_num=part_num, - cb=create_cb(fname, offset, source_size)) - - if len(multipart.get_all_parts()) != chunk_count: - raise Exception("Couldn't upload all file parts") - - multipart.complete_upload() - self._unlink_upload_tracker(fname) - - def push(self, path): - """ push, aws version """ - - aws_key = self.cache_file_key(path) - bucket = self._get_bucket_aws(self.storage_bucket) - key = bucket.get_key(aws_key) - if key: - Logger.debug('File already uploaded to the cloud. Checksum validation...') - - if self._cmp_checksum(key, path): - Logger.debug('File checksum matches. No uploading is needed.') - return path - - Logger.debug('Checksum miss-match. Re-uploading is required.') - - key = bucket.new_key(aws_key) - - try: - self._push_multipart(key, path) - except Exception as exc: - Logger.error('Failed to upload "{}": {}'.format(path, exc)) - return None - - progress.finish_target(os.path.basename(path)) - - return path - - def _status(self, path): - aws_key = self.cache_file_key(path) - bucket = self._get_bucket_aws(self.storage_bucket) - key = bucket.get_key(aws_key) - - remote_exists = key is not None - local_exists = os.path.exists(path) - diff = None - if remote_exists and local_exists: - diff = self._cmp_checksum(key, path) - - return (local_exists, remote_exists, diff) - - def remove(self, path): - aws_file_name = self.cache_file_key(path) - - Logger.debug(u'[Cmd-Remove] Remove from cloud {}.'.format(aws_file_name)) - - if not self._aws_creds.access_key_id or not self._aws_creds.secret_access_key: - Logger.debug('[Cmd-Remove] Unable to check cache file in the cloud') - return - conn = S3Connection(self._aws_creds.access_key_id, self._aws_creds.secret_access_key) - bucket_name = self.storage_bucket - bucket = conn.lookup(bucket_name) - if bucket: - key = bucket.get_key(aws_file_name) - if not key: - Logger.warn('[Cmd-Remove] S3 remove warning: ' - 'file "{}" does not exist in S3'.format(aws_file_name)) - else: - key.delete() - Logger.info('[Cmd-Remove] File "{}" was removed from S3'.format(aws_file_name)) - - -class DataCloudGCP(DataCloudBase): - """ DataCloud class for Google Cloud Platform """ - @property - def gc_project_name(self): - """ - Get project name from config. - """ - return self._cloud_settings.cloud_config.get('ProjectName', None) - - def sanity_check(self): - project = self.gc_project_name - if project is None or len(project) < 1: - raise ConfigError('can\'t read google cloud project name. ' - 'Please set ProjectName in section GC.') - - def _get_bucket_gc(self, storage_bucket): - """ get a bucket object, gc """ - client = gc.Client(project=self.gc_project_name) - bucket = client.bucket(storage_bucket) - if not bucket.exists(): - raise DataCloudError('sync up: google cloud bucket {} ' - 'doesn\'t exist'.format(self.storage_bucket)) - return bucket - - @staticmethod - def _cmp_checksum(blob, fname): - """ - Verify local and remote checksums. - """ - md5 = file_md5(fname)[1] - b64_encoded_md5 = base64.b64encode(md5).decode() if md5 else None - - if blob.md5_hash == b64_encoded_md5: - return True - - return False - - def _import(self, bucket_name, key, fname): - - bucket = self._get_bucket_gc(bucket_name) - - name = os.path.basename(fname) - tmp_file = self.tmp_file(fname) - - blob = bucket.get_blob(key) - if not blob: - Logger.error('File "{}" does not exist in the cloud'.format(key)) - return None - - if self._cmp_checksum(blob, fname): - Logger.debug('File "{}" matches with "{}".'.format(fname, key)) - return fname - - Logger.debug('Downloading cache file from gc "{}/{}"'.format(bucket.name, key)) - - # 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) - - try: - blob.download_to_filename(tmp_file) - except Exception as exc: - Logger.error('Failed to download "{}": {}'.format(key, exc)) - return None - - os.rename(tmp_file, fname) - - progress.finish_target(name) - - Logger.debug('Downloading completed') - - return fname - - def push(self, path): - """ push, gcp version """ - - bucket = self._get_bucket_gc(self.storage_bucket) - blob_name = self.cache_file_key(path) - name = os.path.basename(path) - - blob = bucket.get_blob(blob_name) - if blob is not None and blob.exists(): - if self._cmp_checksum(blob, path): - Logger.debug('checksum %s matches. Skipping upload' % path) - return path - Logger.debug('checksum %s mismatch. re-uploading' % path) - - # same as in _import - progress.update_target(name, 0, None) - - blob = bucket.blob(blob_name) - blob.upload_from_filename(path) - - progress.finish_target(name) - Logger.debug('uploading %s completed' % path) - - return path - - def _status(self, path): - """ status, gcp version """ - - bucket = self._get_bucket_gc(self.storage_bucket) - blob_name = self.cache_file_key(path) - blob = bucket.get_blob(blob_name) - - remote_exists = blob is not None and blob.exists() - local_exists = os.path.exists(path) - diff = None - if remote_exists and local_exists: - diff = self._cmp_checksum(blob, path) - - return (local_exists, remote_exists, diff) - - def remove(self, item): - raise Exception('NOT IMPLEMENTED YET') - - -class DataCloud(object): - """ Generic class to do initial config parsing and redirect to proper DataCloud methods """ - - CLOUD_MAP = { - 'AWS' : DataCloudAWS, - 'GCP' : DataCloudGCP, - 'HTTP' : DataCloudHTTP, - 'LOCAL' : DataCloudLOCAL, - } - - SCHEME_MAP = { - 's3' : 'AWS', - 'http' : 'HTTP', - 'https' : 'HTTP', - 'ftp' : 'HTTP', - 'gs' : 'GCP', - '' : 'LOCAL', - } - - def __init__(self, config): - self._config = config - - cloud_type = self._config['Global'].get('Cloud', '').strip().upper() - if cloud_type not in self.CLOUD_MAP.keys(): - raise ConfigError('Wrong cloud type %s specified' % cloud_type) - - if cloud_type not in self._config.keys(): - raise ConfigError('Can\'t find cloud section \'[%s]\' in config' % cloud_type) - - cloud_settings = self.get_cloud_settings(self._config, - cloud_type) - - self.typ = cloud_type - self._cloud = self.CLOUD_MAP[cloud_type](cloud_settings) - - self.sanity_check() - - @staticmethod - def get_cloud_settings(config, cloud_type): - """ - Obtain cloud settings from config. - """ - if cloud_type not in config.keys(): - cloud_config = None - else: - cloud_config = config[cloud_type] - global_storage_path = config['Global'].get('StoragePath', None) - cloud_settings = CloudSettings(global_storage_path, cloud_config) - return cloud_settings - - def sanity_check(self): - """ sanity check a config - - check that we have a cloud and storagePath - if aws, check can read credentials - if google, check ProjectName - - Returns: - (T,) if good - (F, issues) if bad - """ - key = 'Cloud' - if key.lower() not in [k.lower() for k in self._config['Global'].keys()] or len(self._config['Global'][key]) < 1: - raise ConfigError('Please set %s in section Global in config file' % key) - - # now that a cloud is chosen, can check StoragePath - storage_path = self._cloud.storage_path - if storage_path is None or len(storage_path) == 0: - raise ConfigError('Please set StoragePath = bucket/{optional path} ' - 'in config file in a cloud specific section') - - self._cloud.sanity_check() - - def _map_targets(self, func, targets, jobs): - """ - Process targets as data items in parallel. - """ - return map_progress(func, targets, jobs) - - def sync(self, targets, jobs=1): - """ - Sync data items in a cloud-agnostic way. - """ - return self._map_targets(self._cloud.sync, targets, jobs) - - def push(self, targets, jobs=1): - """ - Push data items in a cloud-agnostic way. - """ - return self._map_targets(self._cloud.push, targets, jobs) - - def pull(self, targets, jobs=1): - """ - Pull data items in a cloud-agnostic way. - """ - return self._map_targets(self._cloud.pull, targets, jobs) - - def status(self, targets, jobs=1): - """ - Check status of data items in a cloud-agnostic way. - """ - return self._map_targets(self._cloud.status, targets, jobs) diff --git a/dvc/project.py b/dvc/project.py index 61b7c3585..099b240fa 100644 --- a/dvc/project.py +++ b/dvc/project.py @@ -9,7 +9,7 @@ from dvc.state import State from dvc.lock import Lock from dvc.scm import SCM from dvc.cache import Cache -from dvc.data_cloud import DataCloud +from dvc.cloud.data_cloud import DataCloud class PipelineError(DvcException): diff --git a/dvc/stage.py b/dvc/stage.py index 1de91b039..4bd12d16e 100644 --- a/dvc/stage.py +++ b/dvc/stage.py @@ -4,7 +4,7 @@ import yaml import itertools from dvc.system import System -from dvc.data_cloud import file_md5 +from dvc.utils import file_md5 from dvc.exceptions import DvcException from dvc.executor import Executor diff --git a/dvc/utils.py b/dvc/utils.py index a0182732b..85729559b 100644 --- a/dvc/utils.py +++ b/dvc/utils.py @@ -5,6 +5,7 @@ import os import re import stat import shutil +import hashlib from multiprocessing.pool import ThreadPool @@ -15,6 +16,18 @@ from dvc.logger import Logger LOCAL_CHUNK_SIZE = 1024*1024 +def file_md5(fname): + """ get the (md5 hexdigest, md5 digest) of a file """ + if os.path.exists(fname): + hash_md5 = hashlib.md5() + with open(fname, "rb") as fobj: + for chunk in iter(lambda: fobj.read(1024*1000), b""): + hash_md5.update(chunk) + return (hash_md5.hexdigest(), hash_md5.digest()) + else: + return (None, None) + + def cached_property(func): '''A decorator for caching properties in classes.''' def get(self): diff --git a/tests/test_add.py b/tests/test_add.py index 17a51f172..90fee66ea 100644 --- a/tests/test_add.py +++ b/tests/test_add.py @@ -3,7 +3,7 @@ import shutil import filecmp from dvc.main import main -from dvc.data_cloud import file_md5 +from dvc.utils import file_md5 from dvc.stage import Stage, CmdOutputNoCacheError, CmdOutputOutsideOfRepoError from dvc.stage import CmdOutputDoesNotExistError, CmdOutputIsNotFileError from dvc.stage import CmdOutputAlreadyTrackedError diff --git a/tests/test_config.py b/tests/test_config.py index 46f6b44df..0af5b3a58 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -16,7 +16,7 @@ except ImportError: # Python 3 builtin_module_name = 'builtins' -from dvc.data_cloud import DataCloud +from dvc.cloud.data_cloud import DataCloud class TestConfigTest(TestCase): diff --git a/tests/test_data_cloud.py b/tests/test_data_cloud.py index e08d3514d..b3e3d09ee 100644 --- a/tests/test_data_cloud.py +++ b/tests/test_data_cloud.py @@ -2,8 +2,8 @@ import os import uuid from dvc.config import ConfigError -from dvc.data_cloud import DataCloud, DataCloudAWS, DataCloudLOCAL, DataCloudHTTP, DataCloudGCP -from dvc.data_cloud import STATUS_UNKNOWN, STATUS_OK, STATUS_MODIFIED, STATUS_NEW, STATUS_DELETED +from dvc.cloud.data_cloud import DataCloud, DataCloudAWS, DataCloudLOCAL, DataCloudHTTP, DataCloudGCP +from dvc.cloud.base import STATUS_UNKNOWN, STATUS_OK, STATUS_MODIFIED, STATUS_NEW, STATUS_DELETED from dvc.cloud.instance_manager import CloudSettings from tests.basic_env import TestDvc diff --git a/tests/test_gc.py b/tests/test_gc.py index e7e042dea..a4d7e3ae9 100644 --- a/tests/test_gc.py +++ b/tests/test_gc.py @@ -3,7 +3,7 @@ import shutil import filecmp from dvc.main import main -from dvc.data_cloud import file_md5 +from dvc.utils import file_md5 from dvc.stage import Stage, CmdOutputNoCacheError, CmdOutputOutsideOfRepoError from dvc.stage import CmdOutputDoesNotExistError, CmdOutputIsNotFileError from dvc.project import StageNotFoundError diff --git a/tests/test_repro.py b/tests/test_repro.py index 81a64cbf4..d90bb7bfe 100644 --- a/tests/test_repro.py +++ b/tests/test_repro.py @@ -6,7 +6,7 @@ import filecmp from dvc.main import main from dvc.command.repro import CmdRepro from dvc.project import ReproductionError -from dvc.data_cloud import file_md5 +from dvc.utils import file_md5 from tests.basic_env import TestDvc diff --git a/tests/test_run.py b/tests/test_run.py index 1294af44c..765aa1c53 100644 --- a/tests/test_run.py +++ b/tests/test_run.py @@ -2,7 +2,7 @@ import os import filecmp from dvc.main import main -from dvc.data_cloud import file_md5 +from dvc.utils import file_md5 from dvc.stage import Stage from dvc.command.run import CmdRun