From 2e78bb2a6961223aa398fcff3db6e9d4240d04ad Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 28 May 2018 23:34:56 +0300 Subject: [PATCH 01/18] deps: s3/gs: check if key exists Signed-off-by: Ruslan Kuprieiev --- dvc/dependency/gs.py | 14 ++++++++------ dvc/dependency/s3.py | 23 +++++++++++++++++------ 2 files changed, 25 insertions(+), 12 deletions(-) diff --git a/dvc/dependency/gs.py b/dvc/dependency/gs.py index 448d6ef8d..e297f452e 100644 --- a/dvc/dependency/gs.py +++ b/dvc/dependency/gs.py @@ -12,12 +12,14 @@ from dvc.cloud.gcp import DataCloudGCP class DependencyGS(DependencyS3): REGEX = DataCloudGCP.REGEX + @property + def client(self): + return gc.Client() + def get_etag(self): - o = urlparse(self.path) - bucket_name = o.netloc - key = o.path.lstrip('/') + bucket = self.client.bucket(self.bucket) + blob = bucket.get_blob(self.key) + if not blob: + return None - client = gc.Client() - bucket = client.bucket(bucket_name) - blob = bucket.get_blob(key) return blob.etag diff --git a/dvc/dependency/s3.py b/dvc/dependency/s3.py index 9b8bfb971..19ebe0cfe 100644 --- a/dvc/dependency/s3.py +++ b/dvc/dependency/s3.py @@ -18,14 +18,25 @@ class DependencyS3(DependencyBase): super(DependencyS3, self).__init__(stage, path) self.etag = etag - def get_etag(self): - o = urlparse(self.path) - bucket = o.netloc - key = o.path.lstrip('/') + @property + def bucket(self): + return urlparse(self.path).netloc + @property + def key(self): + return urlparse(self.path).path.lstrip('/') + + @property + def s3(self): session = boto3.Session() - s3 = session.resource('s3') - obj = s3.Object(bucket, key).get() + return session.resource('s3') + + def get_etag(self): + try: + obj = self.s3.Object(self.bucket, self.key).get() + except Exception: + return None + return obj['ETag'].strip('"') def changed(self): From e0a42f1265e85efa941d2a35fc91c5d80ddc5d26 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 28 May 2018 23:37:05 +0300 Subject: [PATCH 02/18] outs: support gs/s3 as an output without cache Signed-off-by: Ruslan Kuprieiev --- dvc/output/gs.py | 68 +++++++++++++++++++++++++++++++++-- dvc/output/s3.py | 63 +++++++++++++++++++++++++++++++-- tests/test_repro.py | 86 +++++++++++++++++++++++++++------------------ 3 files changed, 179 insertions(+), 38 deletions(-) diff --git a/dvc/output/gs.py b/dvc/output/gs.py index 0cfa59b9c..e70e9ab79 100644 --- a/dvc/output/gs.py +++ b/dvc/output/gs.py @@ -1,5 +1,69 @@ from dvc.dependency.gs import DependencyGS + class OutputGS(DependencyGS): - def __init__(self, stage, path, etag=None): - raise NotImplemented + PARAM_CACHE = 'cache' + + def __init__(self, stage, path, etag=None, use_cache=True): + super(OutputGS, self).__init__(stage, path, etag=etag) + self.use_cache = use_cache + + @property + def cache(self): + if not self.use_cache: + return None + + raise NotImplementedError + + return self.project.cache.get(self.etag) + + def dumpd(self): + ret = super(OutputGS, self).dumpd() + ret[self.PARAM_CACHE] = self.use_cache + return ret + + @classmethod + def loadd(cls, stage, d): + ret = super(OutputGS, cls).loadd(stage, d) + ret.use_cache = d.get(cls.PARAM_CACHE, True) + return ret + + @classmethod + def loads(cls, stage, s, use_cache=True): + ret = super(OutputGS, cls).loads(stage, s) + ret.use_cache = use_cache + return ret + + @classmethod + def loads_from(cls, stage, s_list, use_cache=True): + return [cls.loads(stage, x, use_cache=use_cache) for x in s_list] + + def changed(self): + if super(OutputGS, self).changed(): + return True + + if self.use_cache and self.project.cache.changed(self.etag): + return True + + return False + + def checkout(self): + if not self.use_cache: + return + + raise NotImplementedError + + def save(self): + super(OutputGS, self).save() + + if not self.use_cache: + return + + raise NotImplementedError + + def remove(self): + blob = self.client.bucket(self.bucket).get_blob(self.key) + if not blob: + return + + blob.delete() diff --git a/dvc/output/s3.py b/dvc/output/s3.py index b0971dc66..84cf36a52 100644 --- a/dvc/output/s3.py +++ b/dvc/output/s3.py @@ -2,5 +2,64 @@ from dvc.dependency.s3 import DependencyS3 class OutputS3(DependencyS3): - def __init__(self, stage, path, etag=None): - raise NotImplemented + PARAM_CACHE = 'cache' + + def __init__(self, stage, path, etag=None, use_cache=True): + super(OutputS3, self).__init__(stage, path, etag=etag) + self.use_cache = use_cache + + @property + def cache(self): + if not self.use_cache: + return None + + raise NotImplementedError + + return self.project.cache.get(self.etag) + + def dumpd(self): + ret = super(OutputS3, self).dumpd() + ret[self.PARAM_CACHE] = self.use_cache + return ret + + @classmethod + def loadd(cls, stage, d): + ret = super(OutputS3, cls).loadd(stage, d) + ret.use_cache = d.get(cls.PARAM_CACHE, True) + return ret + + @classmethod + def loads(cls, stage, s, use_cache=True): + ret = super(OutputS3, cls).loads(stage, s) + ret.use_cache = use_cache + return ret + + @classmethod + def loads_from(cls, stage, s_list, use_cache=True): + return [cls.loads(stage, x, use_cache=use_cache) for x in s_list] + + def changed(self): + if super(OutputS3, self).changed(): + return True + + if self.use_cache and self.project.cache.changed(self.etag): + return True + + return False + + def checkout(self): + if not self.use_cache: + return + + raise NotImplementedError + + def save(self): + super(OutputS3, self).save() + + if not self.use_cache: + return + + raise NotImplementedError + + def remove(self): + self.s3.Object(self.bucket, self.key).delete() diff --git a/tests/test_repro.py b/tests/test_repro.py index 5c13660e0..b5f5f9cd8 100644 --- a/tests/test_repro.py +++ b/tests/test_repro.py @@ -186,56 +186,74 @@ class TestCmdRepro(TestRepro): self.assertNotEqual(ret, 0) -class TestReproDependencyS3(TestDvc): +class TestReproExternalBase(TestDvc): + def should_test(self): + return False + def test(self): - if not _should_test_aws(): + if not self.should_test(): return - key = str(uuid.uuid4()) + '/' + self.FOO - path = 's3://' + TEST_AWS_REPO_BUCKET + '/' + key + foo_key = str(uuid.uuid4()) + '/' + self.FOO + bar_key = str(uuid.uuid4()) + '/' + self.BAR - s3 = boto3.resource('s3') + foo_path = self.schema + self.bucket + '/' + foo_key + bar_path = self.schema + self.bucket + '/' + bar_key - with open(self.FOO, 'rb') as fd: - s3.Bucket(TEST_AWS_REPO_BUCKET).put_object(Key=key, Body=fd.read()) + self.write(self.bucket, foo_key, 'foo') - stage = self.dvc.run(outs=[self.FOO], - deps=[path], - cmd='aws s3 cp {} {}'.format(path, self.FOO)) + stage = self.dvc.run(outs_no_cache=[bar_path], + deps=[foo_path], + cmd='{} {} {}'.format(self.cmd, foo_path, bar_path)) - with open(self.BAR, 'rb') as fd: - s3.Bucket(TEST_AWS_REPO_BUCKET).put_object(Key=key, Body=fd.read()) + self.write(self.bucket, foo_key, 'bar') stages = self.dvc.reproduce(stage.path) self.assertEqual(len(stages), 1) - self.assertTrue(filecmp.cmp(self.FOO, self.BAR)) -class TestReproDependencyGS(TestDvc): - def test(self): - if not _should_test_gcp(): - return +class TestReproExternalS3(TestReproExternalBase): + def should_test(self): + return _should_test_aws() + @property + def schema(self): + return 's3://' + + @property + def bucket(self): + return TEST_AWS_REPO_BUCKET + + @property + def cmd(self): + return 'aws s3 cp' + + def write(self, bucket, key, body): + s3 = boto3.resource('s3') + s3.Bucket(bucket).put_object(Key=key, Body=body) + + +class TestReproExternalGS(TestReproExternalBase): + def should_test(self): #FIXME enable if os.getenv('CI'): - return + return False + return _should_test_gcp() - key = str(uuid.uuid4()) + '/' + self.FOO - path = 'gs://' + TEST_GCP_REPO_BUCKET + '/' + key + @property + def schema(self): + return 'gs://' + @property + def bucket(self): + return TEST_GCP_REPO_BUCKET + + @property + def cmd(self): + return 'gsutil cp' + + def write(self, bucket, key, body): client = gc.Client() - bucket = client.bucket(TEST_GCP_REPO_BUCKET) - - bucket.blob(key).upload_from_filename(self.FOO) - - stage = self.dvc.run(outs=[self.FOO], - deps=[path], - cmd='gsutil cp {} {}'.format(path, self.FOO)) - - bucket.blob(key).upload_from_filename(self.BAR) - - stages = self.dvc.reproduce(stage.path) - - self.assertEqual(len(stages), 1) - self.assertTrue(filecmp.cmp(self.FOO, self.BAR)) + bucket = client.bucket(bucket) + bucket.blob(key).upload_from_string(body) From f3b2497fb9b64324adc2be99d336e35ebb086318 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Tue, 29 May 2018 01:02:22 +0300 Subject: [PATCH 03/18] deps/outs: remove load{s,d}_from methods Signed-off-by: Ruslan Kuprieiev --- dvc/dependency/base.py | 4 ---- dvc/dependency/local.py | 8 -------- dvc/dependency/s3.py | 4 ---- dvc/output/gs.py | 4 ---- dvc/output/local.py | 4 ---- dvc/output/s3.py | 4 ---- 6 files changed, 28 deletions(-) diff --git a/dvc/dependency/base.py b/dvc/dependency/base.py index 15ca934cf..017a2b64d 100644 --- a/dvc/dependency/base.py +++ b/dvc/dependency/base.py @@ -65,7 +65,3 @@ class DependencyBase(object): @classmethod def loads(cls, stage, s): return cls(stage, s) - - @classmethod - def loads_from(cls, stage, s_list): - return [cls.loads(stage, x) for x in s_list] diff --git a/dvc/dependency/local.py b/dvc/dependency/local.py index b168bf5c3..4c413dfbd 100644 --- a/dvc/dependency/local.py +++ b/dvc/dependency/local.py @@ -92,14 +92,6 @@ class DependencyLOCAL(DependencyBase): md5 = d.get(cls.PARAM_MD5, None) return cls(stage, path, md5=md5) - @classmethod - def loadd_from(cls, stage, d_list): - return [cls.loadd(stage, x) for x in d_list] - @classmethod def loads(cls, stage, s): return cls(stage, os.path.join(stage.cwd, s), md5=None) - - @classmethod - def loads_from(cls, stage, s_list): - return [cls.loads(stage, x) for x in s_list] diff --git a/dvc/dependency/s3.py b/dvc/dependency/s3.py index 19ebe0cfe..17f9dfa1e 100644 --- a/dvc/dependency/s3.py +++ b/dvc/dependency/s3.py @@ -58,7 +58,3 @@ class DependencyS3(DependencyBase): @classmethod def loads(cls, stage, s): return cls(stage, s, etag=None) - - @classmethod - def loads_from(cls, stage, s_list): - return [cls.loads(stage, x) for x in s_list] diff --git a/dvc/output/gs.py b/dvc/output/gs.py index e70e9ab79..d69d31d10 100644 --- a/dvc/output/gs.py +++ b/dvc/output/gs.py @@ -34,10 +34,6 @@ class OutputGS(DependencyGS): ret.use_cache = use_cache return ret - @classmethod - def loads_from(cls, stage, s_list, use_cache=True): - return [cls.loads(stage, x, use_cache=use_cache) for x in s_list] - def changed(self): if super(OutputGS, self).changed(): return True diff --git a/dvc/output/local.py b/dvc/output/local.py index a21e44ebf..957b319ff 100644 --- a/dvc/output/local.py +++ b/dvc/output/local.py @@ -47,10 +47,6 @@ class OutputLOCAL(DependencyLOCAL): ret.use_cache = use_cache return ret - @classmethod - def loads_from(cls, stage, s_list, use_cache=False): - return [cls.loads(stage, x, use_cache=use_cache) for x in s_list] - def changed(self): if super(OutputLOCAL, self).changed(): return True diff --git a/dvc/output/s3.py b/dvc/output/s3.py index 84cf36a52..0216010a5 100644 --- a/dvc/output/s3.py +++ b/dvc/output/s3.py @@ -34,10 +34,6 @@ class OutputS3(DependencyS3): ret.use_cache = use_cache return ret - @classmethod - def loads_from(cls, stage, s_list, use_cache=True): - return [cls.loads(stage, x, use_cache=use_cache) for x in s_list] - def changed(self): if super(OutputS3, self).changed(): return True From fbfdecd1de6d7868ab6c7bc94e6c3f8aeeaa0a8a Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Tue, 29 May 2018 01:17:07 +0300 Subject: [PATCH 04/18] outs/deps: remove type-specific SCHEMAs Signed-off-by: Ruslan Kuprieiev --- dvc/dependency/local.py | 5 ----- dvc/output/local.py | 3 --- 2 files changed, 8 deletions(-) diff --git a/dvc/dependency/local.py b/dvc/dependency/local.py index 4c413dfbd..708be27ea 100644 --- a/dvc/dependency/local.py +++ b/dvc/dependency/local.py @@ -19,11 +19,6 @@ class DependencyLOCAL(DependencyBase): DoesNotExistError = DependencyDoesNotExistError IsNotFileOrDirError = DependencyIsNotFileOrDirError - SCHEMA = { - PARAM_PATH: str, - schema.Optional(PARAM_MD5): schema.Or(str, None), - } - def __init__(self, stage, path, md5=None): self.stage = stage self.project = stage.project diff --git a/dvc/output/local.py b/dvc/output/local.py index 957b319ff..6b8b080b3 100644 --- a/dvc/output/local.py +++ b/dvc/output/local.py @@ -16,9 +16,6 @@ class OutputLOCAL(DependencyLOCAL): DoesNotExistError = OutputDoesNotExistError IsNotFileOrDirError = OutputIsNotFileOrDirError - SCHEMA = DependencyLOCAL.SCHEMA - SCHEMA[schema.Optional(PARAM_CACHE)] = bool - def __init__(self, stage, path, md5=None, use_cache=True): super(OutputLOCAL, self).__init__(stage, path, md5=md5) self.use_cache = use_cache From 5c1748c15c43c04aae5746b3fa879ec26044fbf0 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Tue, 29 May 2018 01:55:42 +0300 Subject: [PATCH 05/18] outs/deps: remove loads method Signed-off-by: Ruslan Kuprieiev --- dvc/dependency/__init__.py | 2 +- dvc/dependency/base.py | 4 ---- dvc/dependency/local.py | 4 ---- dvc/dependency/s3.py | 4 ---- dvc/output/__init__.py | 2 +- dvc/output/gs.py | 6 ------ dvc/output/local.py | 6 ------ dvc/output/s3.py | 6 ------ 8 files changed, 2 insertions(+), 32 deletions(-) diff --git a/dvc/dependency/__init__.py b/dvc/dependency/__init__.py index 131ee2dc4..ae08a6eca 100644 --- a/dvc/dependency/__init__.py +++ b/dvc/dependency/__init__.py @@ -35,5 +35,5 @@ def loadd_from(stage, d_list): def loads_from(stage, s_list): ret = [] for s in s_list: - ret.append(_get(s).loads(stage, s)) + ret.append(_get(s)(stage, s)) return ret diff --git a/dvc/dependency/base.py b/dvc/dependency/base.py index 017a2b64d..171c9819a 100644 --- a/dvc/dependency/base.py +++ b/dvc/dependency/base.py @@ -61,7 +61,3 @@ class DependencyBase(object): def loadd(cls, stage, d): path = d[cls.PARAM_PATH] return cls(stage, path) - - @classmethod - def loads(cls, stage, s): - return cls(stage, s) diff --git a/dvc/dependency/local.py b/dvc/dependency/local.py index 708be27ea..97dc37c1c 100644 --- a/dvc/dependency/local.py +++ b/dvc/dependency/local.py @@ -86,7 +86,3 @@ class DependencyLOCAL(DependencyBase): path = os.path.normpath(path) md5 = d.get(cls.PARAM_MD5, None) return cls(stage, path, md5=md5) - - @classmethod - def loads(cls, stage, s): - return cls(stage, os.path.join(stage.cwd, s), md5=None) diff --git a/dvc/dependency/s3.py b/dvc/dependency/s3.py index 17f9dfa1e..522730978 100644 --- a/dvc/dependency/s3.py +++ b/dvc/dependency/s3.py @@ -54,7 +54,3 @@ class DependencyS3(DependencyBase): path = d[cls.PARAM_PATH] etag = d.get(cls.PARAM_ETAG, None) return cls(stage, path, etag=etag) - - @classmethod - def loads(cls, stage, s): - return cls(stage, s, etag=None) diff --git a/dvc/output/__init__.py b/dvc/output/__init__.py index 23dd3b542..5139af482 100644 --- a/dvc/output/__init__.py +++ b/dvc/output/__init__.py @@ -33,5 +33,5 @@ def loadd_from(stage, d_list): def loads_from(stage, s_list, use_cache=False): ret = [] for s in s_list: - ret.append(_get(s).loads(stage, s, use_cache=use_cache)) + ret.append(_get(s)(stage, s, use_cache=use_cache)) return ret diff --git a/dvc/output/gs.py b/dvc/output/gs.py index d69d31d10..3a89e7e7f 100644 --- a/dvc/output/gs.py +++ b/dvc/output/gs.py @@ -28,12 +28,6 @@ class OutputGS(DependencyGS): ret.use_cache = d.get(cls.PARAM_CACHE, True) return ret - @classmethod - def loads(cls, stage, s, use_cache=True): - ret = super(OutputGS, cls).loads(stage, s) - ret.use_cache = use_cache - return ret - def changed(self): if super(OutputGS, self).changed(): return True diff --git a/dvc/output/local.py b/dvc/output/local.py index 6b8b080b3..2f761f547 100644 --- a/dvc/output/local.py +++ b/dvc/output/local.py @@ -38,12 +38,6 @@ class OutputLOCAL(DependencyLOCAL): ret.use_cache = d.get(cls.PARAM_CACHE, True) return ret - @classmethod - def loads(cls, stage, s, use_cache=True): - ret = super(OutputLOCAL, cls).loads(stage, s) - ret.use_cache = use_cache - return ret - def changed(self): if super(OutputLOCAL, self).changed(): return True diff --git a/dvc/output/s3.py b/dvc/output/s3.py index 0216010a5..b5df0feca 100644 --- a/dvc/output/s3.py +++ b/dvc/output/s3.py @@ -28,12 +28,6 @@ class OutputS3(DependencyS3): ret.use_cache = d.get(cls.PARAM_CACHE, True) return ret - @classmethod - def loads(cls, stage, s, use_cache=True): - ret = super(OutputS3, cls).loads(stage, s) - ret.use_cache = use_cache - return ret - def changed(self): if super(OutputS3, self).changed(): return True From 08dcd8490ee4daae9c0f95172e01cc4f66b1e959 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Tue, 29 May 2018 02:09:47 +0300 Subject: [PATCH 06/18] outs/deps: remove loadd Signed-off-by: Ruslan Kuprieiev --- dvc/dependency/__init__.py | 2 +- dvc/dependency/base.py | 5 ----- dvc/dependency/local.py | 15 ++++----------- dvc/dependency/s3.py | 6 ------ dvc/output/__init__.py | 4 ++-- dvc/output/gs.py | 10 ++-------- dvc/output/local.py | 10 ++-------- dvc/output/s3.py | 10 ++-------- 8 files changed, 13 insertions(+), 49 deletions(-) diff --git a/dvc/dependency/__init__.py b/dvc/dependency/__init__.py index ae08a6eca..b513e31dd 100644 --- a/dvc/dependency/__init__.py +++ b/dvc/dependency/__init__.py @@ -28,7 +28,7 @@ def loadd_from(stage, d_list): ret = [] for d in d_list: p = d[DependencyBase.PARAM_PATH] - ret.append(_get(p).loadd(stage, d)) + ret.append(_get(p)(stage, **d)) return ret diff --git a/dvc/dependency/base.py b/dvc/dependency/base.py index 171c9819a..31fd48437 100644 --- a/dvc/dependency/base.py +++ b/dvc/dependency/base.py @@ -56,8 +56,3 @@ class DependencyBase(object): def dumpd(self): return {self.PARAM_PATH: self.path} - - @classmethod - def loadd(cls, stage, d): - path = d[cls.PARAM_PATH] - return cls(stage, path) diff --git a/dvc/dependency/local.py b/dvc/dependency/local.py index 97dc37c1c..847635658 100644 --- a/dvc/dependency/local.py +++ b/dvc/dependency/local.py @@ -22,7 +22,10 @@ class DependencyLOCAL(DependencyBase): def __init__(self, stage, path, md5=None): self.stage = stage self.project = stage.project - self.path = os.path.abspath(os.path.normpath(path)) + if not os.path.isabs(path): + path = self.unixpath(path) + path = os.path.join(stage.cwd, path) + self.path = os.path.normpath(path) self.md5 = md5 @property @@ -76,13 +79,3 @@ class DependencyLOCAL(DependencyBase): return {self.PARAM_PATH: path, self.PARAM_MD5:self.md5} - - @classmethod - def loadd(cls, stage, d): - path = d[cls.PARAM_PATH] - if not os.path.isabs(path): - path = cls.unixpath(path) - path = os.path.join(stage.cwd, path) - path = os.path.normpath(path) - md5 = d.get(cls.PARAM_MD5, None) - return cls(stage, path, md5=md5) diff --git a/dvc/dependency/s3.py b/dvc/dependency/s3.py index 522730978..52974057b 100644 --- a/dvc/dependency/s3.py +++ b/dvc/dependency/s3.py @@ -48,9 +48,3 @@ class DependencyS3(DependencyBase): def dumpd(self): return {self.PARAM_PATH: self.path, self.PARAM_ETAG: self.etag} - - @classmethod - def loadd(cls, stage, d): - path = d[cls.PARAM_PATH] - etag = d.get(cls.PARAM_ETAG, None) - return cls(stage, path, etag=etag) diff --git a/dvc/output/__init__.py b/dvc/output/__init__.py index 5139af482..0153fee5f 100644 --- a/dvc/output/__init__.py +++ b/dvc/output/__init__.py @@ -26,12 +26,12 @@ def loadd_from(stage, d_list): ret = [] for d in d_list: p = d[DependencyBase.PARAM_PATH] - ret.append(_get(p).loadd(stage, d)) + ret.append(_get(p)(stage, **d)) return ret def loads_from(stage, s_list, use_cache=False): ret = [] for s in s_list: - ret.append(_get(s)(stage, s, use_cache=use_cache)) + ret.append(_get(s)(stage, s, cache=use_cache)) return ret diff --git a/dvc/output/gs.py b/dvc/output/gs.py index 3a89e7e7f..fc84d78bb 100644 --- a/dvc/output/gs.py +++ b/dvc/output/gs.py @@ -4,9 +4,9 @@ from dvc.dependency.gs import DependencyGS class OutputGS(DependencyGS): PARAM_CACHE = 'cache' - def __init__(self, stage, path, etag=None, use_cache=True): + def __init__(self, stage, path, etag=None, cache=True): super(OutputGS, self).__init__(stage, path, etag=etag) - self.use_cache = use_cache + self.use_cache = cache @property def cache(self): @@ -22,12 +22,6 @@ class OutputGS(DependencyGS): ret[self.PARAM_CACHE] = self.use_cache return ret - @classmethod - def loadd(cls, stage, d): - ret = super(OutputGS, cls).loadd(stage, d) - ret.use_cache = d.get(cls.PARAM_CACHE, True) - return ret - def changed(self): if super(OutputGS, self).changed(): return True diff --git a/dvc/output/local.py b/dvc/output/local.py index 2f761f547..e9dd2876a 100644 --- a/dvc/output/local.py +++ b/dvc/output/local.py @@ -16,9 +16,9 @@ class OutputLOCAL(DependencyLOCAL): DoesNotExistError = OutputDoesNotExistError IsNotFileOrDirError = OutputIsNotFileOrDirError - def __init__(self, stage, path, md5=None, use_cache=True): + def __init__(self, stage, path, md5=None, cache=True): super(OutputLOCAL, self).__init__(stage, path, md5=md5) - self.use_cache = use_cache + self.use_cache = cache @property def cache(self): @@ -32,12 +32,6 @@ class OutputLOCAL(DependencyLOCAL): ret[self.PARAM_CACHE] = self.use_cache return ret - @classmethod - def loadd(cls, stage, d): - ret = super(OutputLOCAL, cls).loadd(stage, d) - ret.use_cache = d.get(cls.PARAM_CACHE, True) - return ret - def changed(self): if super(OutputLOCAL, self).changed(): return True diff --git a/dvc/output/s3.py b/dvc/output/s3.py index b5df0feca..8c9fd4d29 100644 --- a/dvc/output/s3.py +++ b/dvc/output/s3.py @@ -4,9 +4,9 @@ from dvc.dependency.s3 import DependencyS3 class OutputS3(DependencyS3): PARAM_CACHE = 'cache' - def __init__(self, stage, path, etag=None, use_cache=True): + def __init__(self, stage, path, etag=None, cache=True): super(OutputS3, self).__init__(stage, path, etag=etag) - self.use_cache = use_cache + self.use_cache = cache @property def cache(self): @@ -22,12 +22,6 @@ class OutputS3(DependencyS3): ret[self.PARAM_CACHE] = self.use_cache return ret - @classmethod - def loadd(cls, stage, d): - ret = super(OutputS3, cls).loadd(stage, d) - ret.use_cache = d.get(cls.PARAM_CACHE, True) - return ret - def changed(self): if super(OutputS3, self).changed(): return True From f03ea7a9159d53297ddf49fec97657850db8eef9 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Thu, 31 May 2018 03:52:29 +0300 Subject: [PATCH 07/18] dvc: move link_state from Cache to Project Signed-off-by: Ruslan Kuprieiev --- dvc/cache.py | 6 +++--- dvc/project.py | 12 +++++++----- dvc/remote/base.py | 3 +++ dvc/transport/__init__.py | 21 +++++++++++++++++++++ 4 files changed, 34 insertions(+), 8 deletions(-) create mode 100644 dvc/remote/base.py create mode 100644 dvc/transport/__init__.py diff --git a/dvc/cache.py b/dvc/cache.py index a3c2a17f2..fa6711150 100644 --- a/dvc/cache.py +++ b/dvc/cache.py @@ -2,7 +2,7 @@ import os import json import shutil -from dvc.state import State, LinkState +from dvc.state import State from dvc.system import System from dvc.logger import Logger from dvc.utils import move, remove @@ -21,7 +21,7 @@ class Cache(object): 'reflink': System.reflink, } - def __init__(self, root_dir, dvc_dir, cache_dir=None, cache_type=None): + def __init__(self, root_dir, dvc_dir, link_state=None, cache_dir=None, cache_type=None): self.cache_type = cache_type cache_dir = cache_dir if cache_dir else self.CACHE_DIR @@ -34,7 +34,7 @@ class Cache(object): os.mkdir(self.cache_dir) self.state = State(self.cache_dir) - self.link_state = LinkState(root_dir, dvc_dir) + self.link_state = link_state self.lock = Lock(self.cache_dir, name=self.CACHE_DIR_LOCK) @staticmethod diff --git a/dvc/project.py b/dvc/project.py index eda8851f6..7708c6e2b 100644 --- a/dvc/project.py +++ b/dvc/project.py @@ -11,7 +11,7 @@ from dvc.logger import Logger from dvc.exceptions import DvcException from dvc.stage import Stage from dvc.config import Config -from dvc.state import State +from dvc.state import State, LinkState from dvc.lock import Lock from dvc.scm import SCM, Base from dvc.cache import Cache @@ -47,8 +47,10 @@ class Project(object): self.config = Config(self.dvc_dir) self.scm = SCM(self.root_dir) self.lock = Lock(self.dvc_dir) - self.cache = Cache(self.root_dir, self.dvc_dir, cache_dir=self.config._config[Config.SECTION_CACHE].get(Config.SECTION_CACHE_DIR, None), - cache_type=self.config._config[Config.SECTION_CACHE].get(Config.SECTION_CACHE_TYPE, None)) + self.link_state = LinkState(self.root_dir, self.dvc_dir) + self.cache = Cache(self.root_dir, self.dvc_dir, link_state=self.link_state, + cache_dir=self.config._config[Config.SECTION_CACHE].get(Config.SECTION_CACHE_DIR, None), + cache_type=self.config._config[Config.SECTION_CACHE].get(Config.SECTION_CACHE_TYPE, None)) self.state = State(self.dvc_dir) self.logger = Logger(self.config._config[Config.SECTION_CORE].get(Config.SECTION_CORE_LOGLEVEL, None)) self.cloud = DataCloud(cache=self.cache, config=self.config._config) @@ -94,7 +96,7 @@ class Project(object): return proj def _ignore(self): - l = [self.cache.link_state.state_file, + l = [self.link_state.state_file, self.cache.link_state._lock_file.lock_file, self.state.state_file, self.state._lock_file.lock_file, @@ -191,7 +193,7 @@ class Project(object): raise StageNotFoundError(target) stages = [Stage.load(self, target)] else: - self.cache.link_state.remove_all() + self.link_state.remove_all() stages = self.stages() for stage in stages: diff --git a/dvc/remote/base.py b/dvc/remote/base.py new file mode 100644 index 000000000..31a7bf96e --- /dev/null +++ b/dvc/remote/base.py @@ -0,0 +1,3 @@ +class Remote(object): + def __init__(self + pass diff --git a/dvc/transport/__init__.py b/dvc/transport/__init__.py new file mode 100644 index 000000000..08b401df7 --- /dev/null +++ b/dvc/transport/__init__.py @@ -0,0 +1,21 @@ +def Transport(path): + pass + +def exists(path): + return Transport(path).exists(path) + + +def copy(inp, outp): + return Transport(inp).copy(inp, outp) + + +def is_file(path): + return Transport(path).is_file(path) + + +def is_dir(path): + return Transport(path).is_dir(path) + + +def is_empty(path): + return Transport(path).is_empty(path) From 6f4d0cd74addda01621f970f33ca0251f0a0b977 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Thu, 31 May 2018 04:14:31 +0300 Subject: [PATCH 08/18] cache: use atomic move to avoid using Lock() Signed-off-by: Ruslan Kuprieiev --- dvc/cache.py | 57 ++++++++++++++++++++++++---------------------------- 1 file changed, 26 insertions(+), 31 deletions(-) diff --git a/dvc/cache.py b/dvc/cache.py index fa6711150..dfa120d80 100644 --- a/dvc/cache.py +++ b/dvc/cache.py @@ -1,18 +1,17 @@ import os import json +import uuid import shutil from dvc.state import State from dvc.system import System from dvc.logger import Logger from dvc.utils import move, remove -from dvc.lock import Lock from dvc.exceptions import DvcException class Cache(object): CACHE_DIR = 'cache' - CACHE_DIR_LOCK = 'cache.lock' CACHE_TYPES = ['reflink', 'hardlink', 'symlink', 'copy'] CACHE_TYPE_MAP = { 'copy': shutil.copyfile, @@ -35,25 +34,23 @@ class Cache(object): self.state = State(self.cache_dir) self.link_state = link_state - self.lock = Lock(self.cache_dir, name=self.CACHE_DIR_LOCK) @staticmethod def init(root_dir, dvc_dir, cache_dir=None): return Cache(root_dir, dvc_dir, cache_dir=None) def all(self): - with self.lock: - clist = [] - for entry in os.listdir(self.cache_dir): - subdir = os.path.join(self.cache_dir, entry) - if not os.path.isdir(subdir): - continue + clist = [] + for entry in os.listdir(self.cache_dir): + subdir = os.path.join(self.cache_dir, entry) + if not os.path.isdir(subdir): + continue - for cache in os.listdir(subdir): - path = os.path.join(subdir, cache) - clist.append(path) + for cache in os.listdir(subdir): + path = os.path.join(subdir, cache) + clist.append(path) - return clist + return clist def get(self, md5): if not md5: @@ -65,7 +62,7 @@ class Cache(object): relpath = os.path.relpath(path, self.cache_dir) return os.path.dirname(relpath) + os.path.basename(relpath) - def _changed(self, md5): + def changed(self, md5): cache = self.get(md5) if self.state.changed(cache, md5=md5): if os.path.exists(cache): @@ -75,10 +72,6 @@ class Cache(object): return False - def changed(self, md5): - with self.lock: - return self._changed(md5) - def link(self, src, link): dname = os.path.dirname(link) if not os.path.exists(dname): @@ -145,10 +138,10 @@ class Cache(object): def is_dir_cache(cache): return cache.endswith(State.MD5_DIR_SUFFIX) - def _checkout(self, path, md5): + def checkout(self, path, md5): cache = self.get(md5) - if not cache or not os.path.exists(cache) or self._changed(md5): + if not cache or not os.path.exists(cache) or self.changed(md5): if cache: Logger.warn(u'\'{}({})\': cache file not found'.format(os.path.relpath(cache), os.path.relpath(path))) @@ -172,17 +165,20 @@ class Cache(object): p = os.path.join(path, relpath) self.link(c, p) - def checkout(self, path, md5): - with self.lock: - return self._checkout(path, md5) + def _move(self, inp, outp): + # moving in two stages to make last the move atomic in + # case inp and outp are in different filesystems + tmp = '{}.{}'.format(outp, str(uuid.uuid4())) + move(inp, tmp) + move(tmp, outp) def _save_file(self, path): md5 = self.state.update(path) cache = self.get(md5) - if self._changed(md5): - move(path, cache) + if self.changed(md5): + self._move(path, cache) self.state.update(cache) - self._checkout(path, md5) + self.checkout(path, md5) def _save_dir(self, path): md5 = self.state.update(path) @@ -203,8 +199,7 @@ class Cache(object): json.dump(dir_info, fd, sort_keys=True) def save(self, path): - with self.lock: - if os.path.isdir(path): - self._save_dir(path) - else: - self._save_file(path) + if os.path.isdir(path): + self._save_dir(path) + else: + self._save_file(path) From 88397ef43025ec90d6a82a624ab525aabc43e54e Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sun, 3 Jun 2018 19:45:34 +0300 Subject: [PATCH 09/18] dvc: use Remote concept in existing code Signed-off-by: Ruslan Kuprieiev --- dvc/cache.py | 229 ++++++----------------------------- dvc/cloud/aws.py | 7 +- dvc/cloud/base.py | 21 ++-- dvc/cloud/local.py | 2 +- dvc/cloud/ssh.py | 2 +- dvc/config.py | 14 ++- dvc/dependency/__init__.py | 14 ++- dvc/dependency/base.py | 6 +- dvc/dependency/gs.py | 24 ++-- dvc/dependency/local.py | 41 +++---- dvc/dependency/s3.py | 44 ++----- dvc/output/__init__.py | 9 +- dvc/output/gs.py | 20 +-- dvc/output/local.py | 46 ++++--- dvc/output/s3.py | 19 +-- dvc/project.py | 25 ++-- dvc/remote/__init__.py | 0 dvc/remote/base.py | 11 +- dvc/remote/gs.py | 76 ++++++++++++ dvc/remote/local.py | 242 +++++++++++++++++++++++++++++++++++++ dvc/remote/s3.py | 77 ++++++++++++ dvc/remote/ssh.py | 5 + dvc/stage.py | 2 +- dvc/transport/__init__.py | 21 ---- tests/test_add.py | 2 +- tests/test_cache.py | 17 ++- tests/test_checkout.py | 6 +- tests/test_data_cloud.py | 4 +- tests/test_gc.py | 6 +- tests/test_repro.py | 6 +- tests/test_stage.py | 5 +- 31 files changed, 602 insertions(+), 401 deletions(-) create mode 100644 dvc/remote/__init__.py create mode 100644 dvc/remote/gs.py create mode 100644 dvc/remote/local.py create mode 100644 dvc/remote/s3.py create mode 100644 dvc/remote/ssh.py delete mode 100644 dvc/transport/__init__.py diff --git a/dvc/cache.py b/dvc/cache.py index dfa120d80..64c4204be 100644 --- a/dvc/cache.py +++ b/dvc/cache.py @@ -1,205 +1,44 @@ -import os -import json -import uuid -import shutil - -from dvc.state import State -from dvc.system import System -from dvc.logger import Logger -from dvc.utils import move, remove -from dvc.exceptions import DvcException +from dvc.remote.local import RemoteLOCAL +from dvc.remote.s3 import RemoteS3 +from dvc.remote.gs import RemoteGS +from dvc.remote.ssh import RemoteSSH class Cache(object): - CACHE_DIR = 'cache' - CACHE_TYPES = ['reflink', 'hardlink', 'symlink', 'copy'] - CACHE_TYPE_MAP = { - 'copy': shutil.copyfile, - 'symlink': System.symlink, - 'hardlink': System.hardlink, - 'reflink': System.reflink, - } + def __init__(self, project): + #FIXME + from dvc.config import Config - def __init__(self, root_dir, dvc_dir, link_state=None, cache_dir=None, cache_type=None): - self.cache_type = cache_type + config = project.config._config[Config.SECTION_CACHE] - cache_dir = cache_dir if cache_dir else self.CACHE_DIR - if os.path.isabs(cache_dir): - self.cache_dir = cache_dir + local = config.get(Config.SECTION_CACHE_LOCAL, None) + if local: + sect = project.config._config[Config.SECTION_REMOTE_FMT.format(local)] else: - self.cache_dir = os.path.abspath(os.path.realpath(os.path.join(dvc_dir, cache_dir))) + sect = {} + d = config.get(Config.SECTION_CACHE_DIR, None) + if d: + sect[Config.SECTION_REMOTE_URL] = d + t = config.get(Config.SECTION_CACHE_TYPE, None) + if t: + sect[Config.SECTION_CACHE_TYPE] = t - if not os.path.exists(self.cache_dir): - os.mkdir(self.cache_dir) + self.local = RemoteLOCAL(project, sect) - self.state = State(self.cache_dir) - self.link_state = link_state + self.s3 = None + s3 = config.get(Config.SECTION_CACHE_S3, None) + if s3: + sect = project.config._config[Config.SECTION_REMOTE_FMT.format(s3)] + self.s3 = RemoteS3(project, sect) - @staticmethod - def init(root_dir, dvc_dir, cache_dir=None): - return Cache(root_dir, dvc_dir, cache_dir=None) + self.gs = None + gs = config.get(Config.SECTION_CACHE_GS, None) + if gs: + sect = project.config._config[Config.SECTION_REMOTE_FMT.format(gs)] + self.gs = RemoteGS(project, sect) - def all(self): - clist = [] - for entry in os.listdir(self.cache_dir): - subdir = os.path.join(self.cache_dir, entry) - if not os.path.isdir(subdir): - continue - - for cache in os.listdir(subdir): - path = os.path.join(subdir, cache) - clist.append(path) - - return clist - - def get(self, md5): - if not md5: - return None - - return os.path.join(self.cache_dir, md5[0:2], md5[2:]) - - def path_to_md5(self, path): - relpath = os.path.relpath(path, self.cache_dir) - return os.path.dirname(relpath) + os.path.basename(relpath) - - def changed(self, md5): - cache = self.get(md5) - if self.state.changed(cache, md5=md5): - if os.path.exists(cache): - Logger.warn('Corrupted cache file {}'.format(os.path.relpath(cache))) - remove(cache) - return True - - return False - - def link(self, src, link): - dname = os.path.dirname(link) - if not os.path.exists(dname): - os.makedirs(dname) - - if self.cache_type != None: - types = [self.cache_type] - else: - types = self.CACHE_TYPES - - for typ in types: - try: - self.CACHE_TYPE_MAP[typ](src, link) - self.link_state.update(link) - return - except Exception as exc: - msg = 'Cache type \'{}\' is not supported'.format(typ) - Logger.debug(msg) - if typ == types[-1]: - raise DvcException(msg, cause=exc) - - @staticmethod - def load_dir_cache(path): - if os.path.isabs(path): - relpath = os.path.relpath(path) - else: - relpath = path - - try: - with open(path, 'r') as fd: - d = json.load(fd) - except Exception as exc: - msg = u'Failed to load dir cache \'{}\'' - Logger.error(msg.format(relpath), exc) - return [] - - if not isinstance(d, list): - msg = u'Dir cache file format error \'{}\': skipping the file' - Logger.error(msg.format(relpath)) - return [] - - return d - - @staticmethod - def get_dir_cache(path): - res = {} - d = Cache.load_dir_cache(path) - - for entry in d: - res[entry[State.PARAM_RELPATH]] = entry[State.PARAM_MD5] - - return res - - def dir_cache(self, cache): - res = {} - dir_cache = self.get_dir_cache(cache) - - for relpath, md5 in dir_cache.items(): - res[relpath] = self.get(md5) - - return res - - @staticmethod - def is_dir_cache(cache): - return cache.endswith(State.MD5_DIR_SUFFIX) - - def checkout(self, path, md5): - cache = self.get(md5) - - if not cache or not os.path.exists(cache) or self.changed(md5): - if cache: - Logger.warn(u'\'{}({})\': cache file not found'.format(os.path.relpath(cache), - os.path.relpath(path))) - remove(path) - return - - if os.path.exists(path): - msg = u'Data \'{}\' exists. Removing before checkout' - Logger.debug(msg.format(os.path.relpath(path))) - remove(path) - - msg = u'Checking out \'{}\' with cache \'{}\'' - Logger.debug(msg.format(os.path.relpath(path), os.path.relpath(cache))) - - if not self.is_dir_cache(cache): - self.link(cache, path) - return - - dir_cache = self.dir_cache(cache) - for relpath, c in dir_cache.items(): - p = os.path.join(path, relpath) - self.link(c, p) - - def _move(self, inp, outp): - # moving in two stages to make last the move atomic in - # case inp and outp are in different filesystems - tmp = '{}.{}'.format(outp, str(uuid.uuid4())) - move(inp, tmp) - move(tmp, outp) - - def _save_file(self, path): - md5 = self.state.update(path) - cache = self.get(md5) - if self.changed(md5): - self._move(path, cache) - self.state.update(cache) - self.checkout(path, md5) - - def _save_dir(self, path): - md5 = self.state.update(path) - cache = self.get(md5) - dname = os.path.dirname(cache) - dir_info = self.state.collect_dir(path) - - for entry in dir_info: - relpath = entry[State.PARAM_RELPATH] - p = os.path.join(path, relpath) - - self._save_file(p) - - if not os.path.isdir(dname): - os.makedirs(dname) - - with open(cache, 'w+') as fd: - json.dump(dir_info, fd, sort_keys=True) - - def save(self, path): - if os.path.isdir(path): - self._save_dir(path) - else: - self._save_file(path) +# self.ssh = None +# ssh = config.get(Config.SECTION_CACHE_SSH, None) +# if ssh: +# sect = project.config._config[Config.SECTION_REMOTE_FMT.format(ssh)] +# self.ssh = RemoteSSH(project, sect) diff --git a/dvc/cloud/aws.py b/dvc/cloud/aws.py index 8dd1b2d4a..094a777bb 100644 --- a/dvc/cloud/aws.py +++ b/dvc/cloud/aws.py @@ -4,10 +4,9 @@ import threading import boto3 -from dvc.config import Config from dvc.logger import Logger from dvc.progress import progress -from dvc.cloud.base import DataCloudError, DataCloudBase +from dvc.cloud.base import DataCloudBase def sizeof_fmt(num, suffix='B'): @@ -57,9 +56,13 @@ class DataCloudAWS(DataCloudBase): @property def profile(self): + #FIXME + from dvc.config import Config return self._cloud_settings.cloud_config.get(Config.SECTION_AWS_PROFILE, None) def connect(self): + #FIXME + from dvc.cloud.base import DataCloudError session = boto3.Session(profile_name=self.profile) self.s3 = session.resource('s3') bucket = self.s3.Bucket(self.storage_bucket) diff --git a/dvc/cloud/base.py b/dvc/cloud/base.py index 21346e654..b44446ef6 100644 --- a/dvc/cloud/base.py +++ b/dvc/cloud/base.py @@ -4,8 +4,6 @@ import tempfile from dvc.logger import Logger from dvc.exceptions import DvcException -from dvc.config import Config, ConfigError -from dvc.cache import Cache STATUS_UNKNOWN = 0 @@ -47,6 +45,8 @@ class DataCloudBase(object): @property def url(self): + #FIXME + from dvc.config import Config config = self._cloud_settings.cloud_config return config.get(Config.SECTION_REMOTE_URL, None) @@ -72,6 +72,8 @@ class DataCloudBase(object): Precedence: Storage, then cloud specific """ + #FIXME + from dvc.config import Config if self._cloud_settings.global_storage_path: return self._cloud_settings.global_storage_path @@ -108,7 +110,7 @@ class DataCloudBase(object): def cache_file_key(self, fname): """ Key of a file within the bucket """ - relpath = os.path.relpath(fname, self._cloud_settings.cache.cache_dir) + relpath = os.path.relpath(fname, self._cloud_settings.cache.local.cache_dir) relpath = relpath.replace('\\', '/') return '{}/{}'.format(self.storage_prefix, relpath).strip('/') @@ -130,7 +132,10 @@ class DataCloudBase(object): path, local = arg ret = [path] - if not Cache.is_dir_cache(path): + #FIXME + from dvc.remote.local import RemoteLOCAL + + if not RemoteLOCAL.is_dir_cache(path): return ret if local: @@ -146,15 +151,15 @@ class DataCloudBase(object): self._pull_key(key, tmp, no_progress_bar=True) dir_path = tmp - for relpath, md5 in Cache.get_dir_cache(dir_path).items(): - cache = self._cloud_settings.cache.get(md5) + for relpath, md5 in RemoteLOCAL.get_dir_cache(dir_path).items(): + cache = self._cloud_settings.cache.local.get(md5) ret.append(cache) return ret def _cmp_checksum(self, blob, fname): - md5 = self._cloud_settings.cache.path_to_md5(fname) - if self._cloud_settings.cache.state.changed(fname, md5=md5): + md5 = self._cloud_settings.cache.local.path_to_md5(fname) + if self._cloud_settings.cache.local.state.changed(fname, md5=md5): return False return True diff --git a/dvc/cloud/local.py b/dvc/cloud/local.py index 0a3130596..0d1cabe4a 100644 --- a/dvc/cloud/local.py +++ b/dvc/cloud/local.py @@ -23,7 +23,7 @@ class DataCloudLOCAL(DataCloudBase): REGEX = r'^(?P(/+|.:\\+).*)$' def cache_file_key(self, path): - return os.path.relpath(os.path.abspath(path), self._cloud_settings.cache.cache_dir) + return os.path.relpath(os.path.abspath(path), self._cloud_settings.cache.local.cache_dir) def _get_key(self, path): key_name = self.cache_file_key(path) diff --git a/dvc/cloud/ssh.py b/dvc/cloud/ssh.py index 80f7eed89..91fa2c925 100644 --- a/dvc/cloud/ssh.py +++ b/dvc/cloud/ssh.py @@ -52,7 +52,7 @@ class DataCloudSSH(DataCloudBase): return self.ssh.open_sftp() def cache_file_key(self, path): - relpath = os.path.relpath(os.path.abspath(path), self._cloud_settings.cache.cache_dir) + relpath = os.path.relpath(os.path.abspath(path), self._cloud_settings.cache.local.cache_dir) return relpath.replace('\\', '/') def _isfile_remote(self, path): diff --git a/dvc/config.py b/dvc/config.py index 110a46d38..7de2c5127 100644 --- a/dvc/config.py +++ b/dvc/config.py @@ -5,7 +5,6 @@ import os import schema import configobj -from dvc.cache import Cache from dvc.exceptions import DvcException @@ -32,9 +31,17 @@ class Config(object): SECTION_CACHE = 'cache' SECTION_CACHE_DIR = 'dir' SECTION_CACHE_TYPE = 'type' - SECTION_CACHE_TYPE_SCHEMA = schema.And(schema.Use(str.lower), lambda t: t in Cache.CACHE_TYPE_MAP.keys()) + SECTION_CACHE_TYPE_SCHEMA = schema.And(schema.Use(str.lower), lambda t: t in ('reflink', 'hardlink', 'symlink', 'copy')) + SECTION_CACHE_LOCAL = 'local' + SECTION_CACHE_S3 = 's3' + SECTION_CACHE_GS = 'gs' SECTION_CACHE_SCHEMA = { - schema.Optional(SECTION_CACHE_DIR, default=Cache.CACHE_DIR): str, + schema.Optional(SECTION_CACHE_LOCAL): str, + schema.Optional(SECTION_CACHE_S3): str, + schema.Optional(SECTION_CACHE_GS): str, + + # backward compatibility + schema.Optional(SECTION_CACHE_DIR, default='cache'): str, schema.Optional(SECTION_CACHE_TYPE, default=None): SECTION_CACHE_TYPE_SCHEMA, } @@ -91,6 +98,7 @@ class Config(object): schema.Optional(SECTION_AWS_PROFILE, default='default'): str, schema.Optional(SECTION_AWS_CREDENTIALPATH, default = ''): str, schema.Optional(SECTION_GCP_PROJECTNAME): str, + schema.Optional(SECTION_CACHE_TYPE): SECTION_CACHE_TYPE_SCHEMA, } SCHEMA = { diff --git a/dvc/dependency/__init__.py b/dvc/dependency/__init__.py index b513e31dd..29094aa40 100644 --- a/dvc/dependency/__init__.py +++ b/dvc/dependency/__init__.py @@ -7,13 +7,17 @@ from dvc.dependency.s3 import DependencyS3 from dvc.dependency.gs import DependencyGS from dvc.dependency.local import DependencyLOCAL +#FIXME PARAM_MD5, PARAM_ETAG +#from dvc.remote.local import RemoteLOCAL +#from dvc.remote.s3 import RemoteS3 + DEPS = [DependencyS3, DependencyGS, DependencyLOCAL] SCHEMA = { DependencyBase.PARAM_PATH: str, - schema.Optional(DependencyLOCAL.PARAM_MD5): schema.Or(str, None), - schema.Optional(DependencyS3.PARAM_ETAG): schema.Or(str, None), + schema.Optional('md5'): schema.Or(str, None), + schema.Optional('etag'): schema.Or(str, None), } @@ -27,13 +31,13 @@ def _get(path): def loadd_from(stage, d_list): ret = [] for d in d_list: - p = d[DependencyBase.PARAM_PATH] - ret.append(_get(p)(stage, **d)) + p = d.pop(DependencyBase.PARAM_PATH) + ret.append(_get(p)(stage, p, d)) return ret def loads_from(stage, s_list): ret = [] for s in s_list: - ret.append(_get(s)(stage, s)) + ret.append(_get(s)(stage, s, {})) return ret diff --git a/dvc/dependency/base.py b/dvc/dependency/base.py index 31fd48437..246bc75b0 100644 --- a/dvc/dependency/base.py +++ b/dvc/dependency/base.py @@ -32,6 +32,8 @@ class DependencyBase(object): PARAM_PATH = 'path' def __init__(self, stage, path): + self.stage = stage + self.project = stage.project self.path = path @classmethod @@ -43,7 +45,7 @@ class DependencyBase(object): return cls.match(url) != None def changed(self): - raise NotImplemented + raise NotImplementedError def status(self): if self.changed(): @@ -52,7 +54,7 @@ class DependencyBase(object): return {} def save(self): - raise NotImplemented + raise NotImplementedError def dumpd(self): return {self.PARAM_PATH: self.path} diff --git a/dvc/dependency/gs.py b/dvc/dependency/gs.py index e297f452e..ebe398486 100644 --- a/dvc/dependency/gs.py +++ b/dvc/dependency/gs.py @@ -1,25 +1,21 @@ -from google.cloud import storage as gc - try: from urlparse import urlparse except ImportError: from urllib.parse import urlparse -from dvc.dependency.s3 import DependencyS3 from dvc.cloud.gcp import DataCloudGCP +from dvc.dependency.s3 import DependencyS3 +from dvc.remote.gs import RemoteGS +from dvc.config import Config class DependencyGS(DependencyS3): REGEX = DataCloudGCP.REGEX - @property - def client(self): - return gc.Client() - - def get_etag(self): - bucket = self.client.bucket(self.bucket) - blob = bucket.get_blob(self.key) - if not blob: - return None - - return blob.etag + def __init__(self, stage, path, info=None): + super(DependencyGS, self).__init__(stage, path) + self.info = info + self.remote = RemoteGS(stage.project, {Config.SECTION_REMOTE_URL: '/'}) + self.path_info = {'scheme': 'gs', + 'bucket': urlparse(path).netloc, + 'key': urlparse(path).path.lstrip('/')} diff --git a/dvc/dependency/local.py b/dvc/dependency/local.py index 847635658..f8029dd5b 100644 --- a/dvc/dependency/local.py +++ b/dvc/dependency/local.py @@ -12,45 +12,37 @@ from dvc.cloud.local import DataCloudLOCAL class DependencyLOCAL(DependencyBase): REGEX = r'^(?P(/+|.:\\+)?.*)$' - PARAM_PATH = 'path' - PARAM_MD5 = 'md5' - MD5_DIR_SUFFIX = '.dir' - DoesNotExistError = DependencyDoesNotExistError IsNotFileOrDirError = DependencyIsNotFileOrDirError - def __init__(self, stage, path, md5=None): + def __init__(self, stage, path, info=None): + #FIXME + from dvc.config import Config + from dvc.remote.local import RemoteLOCAL + self.stage = stage self.project = stage.project if not os.path.isabs(path): path = self.unixpath(path) path = os.path.join(stage.cwd, path) self.path = os.path.normpath(path) - self.md5 = md5 + self.info = info + self.remote = RemoteLOCAL(stage.project, + {Config.SECTION_REMOTE_URL: self.project.dvc_dir}) + self.path_info = {'scheme': 'local', + 'path': self.path} @property def rel_path(self): return os.path.relpath(self.path) - def _changed_md5(self): + def changed(self): if not os.path.exists(self.path): return True - return self.project.state.changed(self.path, self.md5) + info = self.remote.save_info(self.path_info) - @staticmethod - def _changed_msg(changed): - if changed: - return 'changed' - return "didn't change" - - def changed(self): - ret = self._changed_md5() - - msg = u'Dependency \'{}\' {}'.format(self.rel_path, self._changed_msg(ret)) - self.project.logger.debug(msg) - - return ret + return self.info != info def save(self): if not os.path.exists(self.path): @@ -63,7 +55,7 @@ class DependencyLOCAL(DependencyBase): (os.path.isdir(self.path) and len(os.listdir(self.path)) == 0): self.project.logger.warn("File/directory '{}' is empty.".format(self.rel_path)) - self.md5 = self.project.state.update(self.path) + self.info = self.remote.save_info(self.path_info) @staticmethod def unixpath(path): @@ -77,5 +69,6 @@ class DependencyLOCAL(DependencyBase): else: path = self.path - return {self.PARAM_PATH: path, - self.PARAM_MD5:self.md5} + info = self.info + info[self.PARAM_PATH] = path + return info diff --git a/dvc/dependency/s3.py b/dvc/dependency/s3.py index 52974057b..78beb4c89 100644 --- a/dvc/dependency/s3.py +++ b/dvc/dependency/s3.py @@ -1,5 +1,3 @@ -import boto3 - try: from urlparse import urlparse except ImportError: @@ -7,44 +5,28 @@ except ImportError: from dvc.dependency.base import DependencyBase from dvc.cloud.aws import DataCloudAWS +from dvc.remote.s3 import RemoteS3 +from dvc.config import Config class DependencyS3(DependencyBase): REGEX = DataCloudAWS.REGEX - PARAM_ETAG = 'etag' - - def __init__(self, stage, path, etag=None): + def __init__(self, stage, path, info=None): super(DependencyS3, self).__init__(stage, path) - self.etag = etag - - @property - def bucket(self): - return urlparse(self.path).netloc - - @property - def key(self): - return urlparse(self.path).path.lstrip('/') - - @property - def s3(self): - session = boto3.Session() - return session.resource('s3') - - def get_etag(self): - try: - obj = self.s3.Object(self.bucket, self.key).get() - except Exception: - return None - - return obj['ETag'].strip('"') + self.info = info + self.remote = RemoteS3(stage.project, {Config.SECTION_REMOTE_URL: '/'}) + self.path_info = {'scheme': 's3', + 'bucket': urlparse(path).netloc, + 'key': urlparse(path).path.lstrip('/')} def changed(self): - return self.etag != self.get_etag() + return self.info != self.remote.save_info(self.path_info) def save(self): - self.etag = self.get_etag() + self.info = self.remote.save_info(self.path_info) def dumpd(self): - return {self.PARAM_PATH: self.path, - self.PARAM_ETAG: self.etag} + ret = self.info + ret[self.PARAM_PATH] = self.path + return ret diff --git a/dvc/output/__init__.py b/dvc/output/__init__.py index 0153fee5f..0b402c83b 100644 --- a/dvc/output/__init__.py +++ b/dvc/output/__init__.py @@ -25,13 +25,14 @@ def _get(path): def loadd_from(stage, d_list): ret = [] for d in d_list: - p = d[DependencyBase.PARAM_PATH] - ret.append(_get(p)(stage, **d)) + p = d.pop(DependencyBase.PARAM_PATH) + cache = d.pop(OutputLOCAL.PARAM_CACHE, True) + ret.append(_get(p)(stage, p, info=d, cache=cache)) return ret -def loads_from(stage, s_list, use_cache=False): +def loads_from(stage, s_list, use_cache=True): ret = [] for s in s_list: - ret.append(_get(s)(stage, s, cache=use_cache)) + ret.append(_get(s)(stage, s, info={}, cache=use_cache)) return ret diff --git a/dvc/output/gs.py b/dvc/output/gs.py index fc84d78bb..f0d037824 100644 --- a/dvc/output/gs.py +++ b/dvc/output/gs.py @@ -4,19 +4,10 @@ from dvc.dependency.gs import DependencyGS class OutputGS(DependencyGS): PARAM_CACHE = 'cache' - def __init__(self, stage, path, etag=None, cache=True): - super(OutputGS, self).__init__(stage, path, etag=etag) + def __init__(self, stage, path, info=None, cache=True): + super(OutputGS, self).__init__(stage, path, info) self.use_cache = cache - @property - def cache(self): - if not self.use_cache: - return None - - raise NotImplementedError - - return self.project.cache.get(self.etag) - def dumpd(self): ret = super(OutputGS, self).dumpd() ret[self.PARAM_CACHE] = self.use_cache @@ -26,7 +17,7 @@ class OutputGS(DependencyGS): if super(OutputGS, self).changed(): return True - if self.use_cache and self.project.cache.changed(self.etag): + if self.use_cache and self.info != self.project.cache.gs.save_info(self.path_info): return True return False @@ -35,7 +26,7 @@ class OutputGS(DependencyGS): if not self.use_cache: return - raise NotImplementedError + self.project.cache.gs.checkout(self.path_info, self.info) def save(self): super(OutputGS, self).save() @@ -43,7 +34,8 @@ class OutputGS(DependencyGS): if not self.use_cache: return - raise NotImplementedError + + self.info = self.project.cache.gs.save(self.path_info) def remove(self): blob = self.client.bucket(self.bucket).get_blob(self.key) diff --git a/dvc/output/local.py b/dvc/output/local.py index e9dd2876a..262e2d642 100644 --- a/dvc/output/local.py +++ b/dvc/output/local.py @@ -13,19 +13,21 @@ from dvc.utils import remove class OutputLOCAL(DependencyLOCAL): PARAM_CACHE = 'cache' - DoesNotExistError = OutputDoesNotExistError - IsNotFileOrDirError = OutputIsNotFileOrDirError + DoesNotExistError = OutputDoesNotExistError + IsNotFileOrDirError = OutputIsNotFileOrDirError - def __init__(self, stage, path, md5=None, cache=True): - super(OutputLOCAL, self).__init__(stage, path, md5=md5) + def __init__(self, stage, path, info=None, cache=True): + super(OutputLOCAL, self).__init__(stage, path, info) self.use_cache = cache @property - def cache(self): - if not self.use_cache: - return None + def md5(self): + #FIXME + return self.info.get(self.project.cache.local.PARAM_MD5, None) - return self.project.cache.get(self.md5) + @property + def cache(self): + return self.project.cache.local.get(self.md5) def dumpd(self): ret = super(OutputLOCAL, self).dumpd() @@ -36,7 +38,7 @@ class OutputLOCAL(DependencyLOCAL): if super(OutputLOCAL, self).changed(): return True - if self.use_cache and self.project.cache.changed(self.md5): + if self.use_cache and self.info != self.project.cache.local.save_info(self.path_info): return True return False @@ -45,31 +47,27 @@ class OutputLOCAL(DependencyLOCAL): if not self.use_cache: return - msg = u'Checking out \'{}\' with cache \'{}\'' - self.project.logger.debug(msg.format(self.rel_path, self.md5)) - if not self.changed(): - msg = u'Data file \'{}\' with cache \'{}\' didn\'t change, skipping checkout.' - self.project.logger.debug(msg.format(self.rel_path, self.md5)) + msg = u'Data file \'{}\' didn\'t change, skipping checkout.' + self.project.logger.debug(msg.format(self.rel_path)) return - self.project.cache.checkout(self.path, self.md5) + self.project.cache.local.checkout(self.path_info, self.info) def save(self): + if not self.changed(): + msg = 'Output \'{}\' didn\'t change. Skipping saving.' + self.project.logger.debug(msg.format(self.rel_path)) + return + super(OutputLOCAL, self).save() if not self.use_cache: + msg = 'Output \'{}\' doesn\'t use cache. Skipping saving.' + self.project.logger.debug(msg.format(self.rel_path)) return - self.project.logger.debug(u'Saving \'{}\' to \'{}\''.format(self.rel_path, self.md5)) - - if self.project.scm.is_tracked(self.path): - raise OutputAlreadyTrackedError(self.rel_path) - - if not self.changed(): - return - - self.project.cache.save(self.path) + self.info = self.project.cache.local.save(self.path_info) def remove(self): remove(self.path) diff --git a/dvc/output/s3.py b/dvc/output/s3.py index 8c9fd4d29..12db2a8b7 100644 --- a/dvc/output/s3.py +++ b/dvc/output/s3.py @@ -4,19 +4,10 @@ from dvc.dependency.s3 import DependencyS3 class OutputS3(DependencyS3): PARAM_CACHE = 'cache' - def __init__(self, stage, path, etag=None, cache=True): - super(OutputS3, self).__init__(stage, path, etag=etag) + def __init__(self, stage, path, info=None, cache=True): + super(OutputS3, self).__init__(stage, path, info) self.use_cache = cache - @property - def cache(self): - if not self.use_cache: - return None - - raise NotImplementedError - - return self.project.cache.get(self.etag) - def dumpd(self): ret = super(OutputS3, self).dumpd() ret[self.PARAM_CACHE] = self.use_cache @@ -26,7 +17,7 @@ class OutputS3(DependencyS3): if super(OutputS3, self).changed(): return True - if self.use_cache and self.project.cache.changed(self.etag): + if self.use_cache and self.info != self.project.cache.s3.save_info(self.path_info): return True return False @@ -35,7 +26,7 @@ class OutputS3(DependencyS3): if not self.use_cache: return - raise NotImplementedError + self.project.cache.s3.checkout(self.path_info, self.info) def save(self): super(OutputS3, self).save() @@ -43,7 +34,7 @@ class OutputS3(DependencyS3): if not self.use_cache: return - raise NotImplementedError + self.info = self.project.cache.s3.save(self.path_info) def remove(self): self.s3.Object(self.bucket, self.key).delete() diff --git a/dvc/project.py b/dvc/project.py index 7708c6e2b..2718974a0 100644 --- a/dvc/project.py +++ b/dvc/project.py @@ -11,7 +11,7 @@ from dvc.logger import Logger from dvc.exceptions import DvcException from dvc.stage import Stage from dvc.config import Config -from dvc.state import State, LinkState +from dvc.state import LinkState from dvc.lock import Lock from dvc.scm import SCM, Base from dvc.cache import Cache @@ -48,11 +48,8 @@ class Project(object): self.scm = SCM(self.root_dir) self.lock = Lock(self.dvc_dir) self.link_state = LinkState(self.root_dir, self.dvc_dir) - self.cache = Cache(self.root_dir, self.dvc_dir, link_state=self.link_state, - cache_dir=self.config._config[Config.SECTION_CACHE].get(Config.SECTION_CACHE_DIR, None), - cache_type=self.config._config[Config.SECTION_CACHE].get(Config.SECTION_CACHE_TYPE, None)) - self.state = State(self.dvc_dir) self.logger = Logger(self.config._config[Config.SECTION_CORE].get(Config.SECTION_CORE_LOGLEVEL, None)) + self.cache = Cache(self) self.cloud = DataCloud(cache=self.cache, config=self.config._config) self.updater = Updater(self.dvc_dir) @@ -85,8 +82,6 @@ class Project(object): os.mkdir(dvc_dir) config = Config.init(dvc_dir) - Cache.init(root_dir, dvc_dir) - State.init(dvc_dir) proj = Project(root_dir) scm.add([config.config_file]) @@ -97,15 +92,13 @@ class Project(object): def _ignore(self): l = [self.link_state.state_file, - self.cache.link_state._lock_file.lock_file, - self.state.state_file, - self.state._lock_file.lock_file, + self.link_state._lock_file.lock_file, self.lock.lock_file, self.config.config_local_file, self.updater.updater_file] - if self.cache.cache_dir.startswith(self.root_dir): - l += [self.cache.cache_dir] + if self.cache.local.cache_dir.startswith(self.root_dir): + l += [self.cache.local.cache_dir] self.scm.ignore_list(l) @@ -212,15 +205,15 @@ class Project(object): if not out.use_cache or not out.cache: continue cache_set |= set([out.cache]) - if self.cache.is_dir_cache(out.cache) and os.path.isfile(out.cache): - dir_cache = self.cache.dir_cache(out.cache) + if self.cache.local.is_dir_cache(out.cache) and os.path.isfile(out.cache): + dir_cache = self.cache.local.dir_cache(out.cache) cache_set |= set(dir_cache.values()) return list(cache_set) def gc(self): clist = self._used_cache() - for cache in self.cache.all(): + for cache in self.cache.local.all(): if cache in clist: continue os.unlink(cache) @@ -262,7 +255,7 @@ class Project(object): cloud.STATUS_NEW: 'new', } - path = os.path.relpath(target, self.cache.cache_dir) + path = os.path.relpath(target, self.cache.local.cache_dir) status[path] = prefix_map[ret] diff --git a/dvc/remote/__init__.py b/dvc/remote/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 31a7bf96e..1906cb356 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -1,3 +1,12 @@ class Remote(object): - def __init__(self + def __init__(self, project, config): + pass + + def save_info(self, path_info): + pass + + def save(self, path_info): + pass + + def checkout(self, path_info, checksum_info): pass diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py new file mode 100644 index 000000000..74b85c4c1 --- /dev/null +++ b/dvc/remote/gs.py @@ -0,0 +1,76 @@ +import posixpath +from google.cloud import storage + +try: + from urlparse import urlparse +except ImportError: + from urllib.parse import urlparse + +from dvc.remote.base import Remote + + +class RemoteGS(Remote): + PARAM_ETAG = 'etag' + + def __init__(self, project, config): + #FIXME + from dvc.config import Config + + self.project = project + self.url = config.get[Config.SECTION_REMOTE_URL] + self.projectname = config.get(Config.SECTION_GCP_PROJECTNAME, None) + + @property + def bucket(self): + return urlparse(self.url).netloc + + @property + def prefix(self): + return urlparse(self.url).path.lstrip('/') + + @property + def gs(self): + return storage.Client() + + def get_etag(self, bucket, key): + blob = self.gs.bucket(bucket).get_blob(key) + if not blob: + return None + + return blob.etag + + def save_info(self, path_info): + if path_info['scheme'] != 'gs': + raise NotImplementedError + + return {self.PARAM_ETAG: self.get_etag(path_info['bucket'], path_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:]) + + 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.bucket, name=dest_key) + + return {PARAM_ETAG: etag} + + def checkout(self, path_info, checksum_info): + if path_info['scheme'] != 'gs': + raise NotImplementedError + + etag = checksum_info.get(PARAM_ETAG, None) + if not etag: + 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)) + + self.gs.bucket(path_info['bucket']).copy_blob(blob, path_info['bucket'], name=path_info['key']) diff --git a/dvc/remote/local.py b/dvc/remote/local.py new file mode 100644 index 000000000..1b3abe78e --- /dev/null +++ b/dvc/remote/local.py @@ -0,0 +1,242 @@ +import os +import uuid +import json +import shutil + +from dvc.system import System +from dvc.remote.base import Remote +from dvc.state import State +from dvc.logger import Logger +from dvc.utils import remove, move +from dvc.output.base import OutputAlreadyTrackedError + + +class RemoteLOCAL(Remote): + PARAM_MD5 = State.PARAM_MD5 + + CACHE_DIR = 'cache' + CACHE_TYPES = ['reflink', 'hardlink', 'symlink', 'copy'] + CACHE_TYPE_MAP = { + 'copy': shutil.copyfile, + 'symlink': System.symlink, + 'hardlink': System.hardlink, + 'reflink': System.reflink, + } + + def __init__(self, project, config): + #FIXME + from dvc.config import Config + + self.project = project + link_state = project.link_state + cache_dir = config.get(Config.SECTION_REMOTE_URL, self.CACHE_DIR) + cache_type = config.get(Config.SECTION_CACHE_TYPE, None) + + self.cache_type = cache_type + + cache_dir = cache_dir if cache_dir else self.CACHE_DIR + if os.path.isabs(cache_dir): + self.cache_dir = cache_dir + else: + self.cache_dir = os.path.abspath(os.path.realpath(os.path.join(project.dvc_dir, cache_dir))) + + if not os.path.exists(self.cache_dir): + os.mkdir(self.cache_dir) + + self.state = State(self.cache_dir) + self.link_state = link_state + + def all(self): + clist = [] + for entry in os.listdir(self.cache_dir): + subdir = os.path.join(self.cache_dir, entry) + if not os.path.isdir(subdir): + continue + + for cache in os.listdir(subdir): + path = os.path.join(subdir, cache) + clist.append(path) + + return clist + + def get(self, md5): + if not md5: + return None + + return os.path.join(self.cache_dir, md5[0:2], md5[2:]) + + def path_to_md5(self, path): + relpath = os.path.relpath(path, self.cache_dir) + return os.path.dirname(relpath) + os.path.basename(relpath) + + def changed(self, md5): + cache = self.get(md5) + if self.state.changed(cache, md5=md5): + if os.path.exists(cache): + Logger.warn('Corrupted cache file {}'.format(os.path.relpath(cache))) + remove(cache) + return True + + return False + + def link(self, src, link): + dname = os.path.dirname(link) + if not os.path.exists(dname): + os.makedirs(dname) + + if self.cache_type != None: + types = [self.cache_type] + else: + types = self.CACHE_TYPES + + for typ in types: + try: + msg = u'Checking out \'{}\' with cache \'{}\'' + Logger.debug(msg.format(os.path.relpath(src), os.path.relpath(link))) + self.CACHE_TYPE_MAP[typ](src, link) + self.link_state.update(link) + return + except Exception as exc: + msg = 'Cache type \'{}\' is not supported'.format(typ) + Logger.debug(msg) + if typ == types[-1]: + raise DvcException(msg, cause=exc) + + @staticmethod + def load_dir_cache(path): + if os.path.isabs(path): + relpath = os.path.relpath(path) + else: + relpath = path + + try: + with open(path, 'r') as fd: + d = json.load(fd) + except Exception as exc: + msg = u'Failed to load dir cache \'{}\'' + Logger.error(msg.format(relpath), exc) + return [] + + if not isinstance(d, list): + msg = u'Dir cache file format error \'{}\': skipping the file' + Logger.error(msg.format(relpath)) + return [] + + return d + + @classmethod + def get_dir_cache(cls, path): + res = {} + + d = cls.load_dir_cache(path) + + for entry in d: + res[entry[State.PARAM_RELPATH]] = entry[State.PARAM_MD5] + + return res + + def dir_cache(self, cache): + res = {} + dir_cache = self.get_dir_cache(cache) + + for relpath, md5 in dir_cache.items(): + res[relpath] = self.get(md5) + + return res + + @staticmethod + def is_dir_cache(cache): + return cache.endswith(State.MD5_DIR_SUFFIX) + + def checkout(self, path_info, checksum_info): + path = path_info['path'] + md5 = checksum_info.get(self.PARAM_MD5, None) + cache = self.get(md5) + + if not cache or not os.path.exists(cache) or self.changed(md5): + if cache: + Logger.warn(u'\'{}({})\': cache file not found'.format(os.path.relpath(cache), + os.path.relpath(path))) + remove(path) + return + + if os.path.exists(path): + msg = u'Data \'{}\' exists. Removing before checkout' + Logger.debug(msg.format(os.path.relpath(path))) + remove(path) + + if not self.is_dir_cache(cache): + self.link(cache, path) + return + + msg = u'Checking out directory \'{}\' with cache \'{}\'' + Logger.debug(msg.format(os.path.relpath(path), os.path.relpath(cache))) + + dir_cache = self.dir_cache(cache) + for relpath, c in dir_cache.items(): + p = os.path.join(path, relpath) + self.link(c, p) + + def _move(self, inp, outp): + # moving in two stages to make last the move atomic in + # case inp and outp are in different filesystems + tmp = '{}.{}'.format(outp, str(uuid.uuid4())) + move(inp, tmp) + move(tmp, outp) + + def _save_file(self, path_info): + path = path_info['path'] + md5 = self.state.update(path) + cache = self.get(md5) + if self.changed(md5): + Logger.debug(u'Saving \'{}\' to \'{}\''.format(os.path.relpath(path), + os.path.relpath(cache))) + self._move(path, cache) + self.state.update(cache) + self.checkout(path_info, {self.PARAM_MD5: md5}) + + return {self.PARAM_MD5: md5} + + def _save_dir(self, path_info): + path = path_info['path'] + md5 = self.state.update(path) + cache = self.get(md5) + dname = os.path.dirname(cache) + dir_info = self.state.collect_dir(path) + + for entry in dir_info: + relpath = entry[State.PARAM_RELPATH] + p = os.path.join(path, relpath) + + self._save_file({'scheme': 'local', 'path': p}) + + if not os.path.isdir(dname): + os.makedirs(dname) + + Logger.debug(u'Saving directory \'{}\' to \'{}\''.format(os.path.relpath(path), + os.path.relpath(cache))) + + with open(cache, 'w+') as fd: + json.dump(dir_info, fd, sort_keys=True) + + return {self.PARAM_MD5: md5} + + def save(self, path_info): + if path_info['scheme'] != 'local': + raise NotImplementedError + + path = path_info['path'] + + if self.project.scm.is_tracked(path): + raise OutputAlreadyTrackedError(os.path.relpath(path)) + + if os.path.isdir(path): + return self._save_dir(path_info) + else: + return self._save_file(path_info) + + def save_info(self, path_info): + if path_info['scheme'] != 'local': + raise NotImplementedError + + return {self.PARAM_MD5: self.state.update(path_info['path'])} diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py new file mode 100644 index 000000000..95cfe03f5 --- /dev/null +++ b/dvc/remote/s3.py @@ -0,0 +1,77 @@ +import boto3 +import posixpath + +try: + from urlparse import urlparse +except ImportError: + from urllib.parse import urlparse + +from dvc.remote.base import Remote + + +class RemoteS3(Remote): + PARAM_ETAG = 'etag' + + def __init__(self, project, config): + #FIXME + from dvc.config import Config + + + self.project = project + self.url = config[Config.SECTION_REMOTE_URL] + self.region = config.get(Config.SECTION_AWS_REGION, None) + self.profile = config.get(Config.SECTION_AWS_PROFILE, None) + self.credentialpath = config.get(Config.SECTION_AWS_CREDENTIALPATH, None) + + @property + def bucket(self): + return urlparse(self.url).netloc + + @property + def prefix(self): + return urlparse(self.url).path.lstrip('/') + + @property + def s3(self): + return boto3.resource('s3') + + def get_etag(self, bucket, key): + try: + obj = self.s3.Object(bucket, key).get() + except Exception: + return None + + return obj['ETag'].strip('"') + + def save_info(self, path_info): + if path_info['scheme'] != 's3': + raise NotImplementedError + + return {self.PARAM_ETAG: self.get_etag(path_info['bucket'], path_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:]) + + source = {'Bucket': path_info['bucket'], + 'Key': path_info['key']} + self.s3.Bucket(self.bucket).copy(source, dest_key) + + return {self.PARAM_ETAG: etag} + + def checkout(self, path_info, checksum_info): + if path_info['scheme'] != 's3': + raise NotImplementedError + + etag = checksum_info.get(self.PARAM_ETAG, None) + if not etag: + return + + key = posixpath.join(self.prefix, etag[0:2], etag[2:]) + source = {'Bucket': self.bucket, + 'Key': key} + + self.s3.Bucket(path_info['bucket']).copy(source, path_info['key']) diff --git a/dvc/remote/ssh.py b/dvc/remote/ssh.py new file mode 100644 index 000000000..3e83c423e --- /dev/null +++ b/dvc/remote/ssh.py @@ -0,0 +1,5 @@ +from dvc.remote.base import Remote + + +class RemoteSSH(Remote): + pass diff --git a/dvc/stage.py b/dvc/stage.py index ca7aaec4d..86e5f5ab7 100644 --- a/dvc/stage.py +++ b/dvc/stage.py @@ -219,7 +219,7 @@ class Stage(object): for out in self.outs: out.save() - if out.use_cache: + if out.use_cache and out.path_info['scheme'] == 'local': self.project.scm.ignore(out.path) def run(self): diff --git a/dvc/transport/__init__.py b/dvc/transport/__init__.py deleted file mode 100644 index 08b401df7..000000000 --- a/dvc/transport/__init__.py +++ /dev/null @@ -1,21 +0,0 @@ -def Transport(path): - pass - -def exists(path): - return Transport(path).exists(path) - - -def copy(inp, outp): - return Transport(inp).copy(inp, outp) - - -def is_file(path): - return Transport(path).is_file(path) - - -def is_dir(path): - return Transport(path).is_dir(path) - - -def is_empty(path): - return Transport(path).is_empty(path) diff --git a/tests/test_add.py b/tests/test_add.py index 9711c0891..b1f0736d3 100644 --- a/tests/test_add.py +++ b/tests/test_add.py @@ -25,7 +25,7 @@ class TestAdd(TestDvc): self.assertEqual(len(stage.outs), 1) self.assertEqual(len(stage.deps), 0) self.assertEqual(stage.cmd, None) - self.assertEqual(stage.outs[0].md5, md5) + self.assertEqual(stage.outs[0].info['md5'], md5) class TestAddNonExistentFile(TestDvc): diff --git a/tests/test_cache.py b/tests/test_cache.py index b917b328a..5a82bc5b4 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -5,6 +5,7 @@ import tempfile from dvc.cache import Cache from dvc.system import System from dvc.main import main +from dvc.remote.local import RemoteLOCAL from tests.basic_env import TestDvc @@ -14,19 +15,23 @@ class TestCache(TestDvc): super(TestCache, self).setUp() self.cache1_md5 = '123' self.cache2_md5 = '234' - self.cache1 = os.path.join(self.dvc.cache.cache_dir, self.cache1_md5[0:2], self.cache1_md5[2:]) - self.cache2 = os.path.join(self.dvc.cache.cache_dir, self.cache2_md5[0:2], self.cache2_md5[2:]) + self.cache1 = os.path.join(self.dvc.cache.local.cache_dir, + self.cache1_md5[0:2], + self.cache1_md5[2:]) + self.cache2 = os.path.join(self.dvc.cache.local.cache_dir, + self.cache2_md5[0:2], + self.cache2_md5[2:]) self.create(self.cache1, '1') self.create(self.cache2, '2') def test_all(self): - flist = Cache(self.dvc.root_dir, self.dvc.dvc_dir).all() + flist = Cache(self.dvc).local.all() self.assertEquals(len(flist), 2) self.assertTrue(self.cache1 in flist) self.assertTrue(self.cache2 in flist) def test_get(self): - cache = Cache(self.dvc.root_dir, self.dvc.dvc_dir).get(self.cache1_md5) + cache = Cache(self.dvc).local.get(self.cache1_md5) self.assertEquals(cache, self.cache1) @@ -38,11 +43,11 @@ class TestCacheLoadBadDirCache(TestDvc): def test(self): fname = 'not-json' self.create(fname, 'not,json') - self._do_test(Cache.load_dir_cache(fname)) + self._do_test(RemoteLOCAL.load_dir_cache(fname)) fname = 'not-list' self.create(fname, '{"a": "b"}') - self._do_test(Cache.load_dir_cache(fname)) + self._do_test(RemoteLOCAL.load_dir_cache(fname)) class TestExternalCacheDir(TestDvc): diff --git a/tests/test_checkout.py b/tests/test_checkout.py index cd93baed8..f39263ac2 100644 --- a/tests/test_checkout.py +++ b/tests/test_checkout.py @@ -9,7 +9,7 @@ from dvc.main import main from tests.basic_env import TestDvc from tests.test_repro import TestRepro from dvc.stage import Stage -from dvc.output.local import OutputLOCAL +from dvc.remote.local import RemoteLOCAL class TestCheckout(TestRepro): @@ -163,8 +163,8 @@ class TestCheckoutMissingMd5InStageFile(TestRepro): with open(self.file1_stage, 'r') as fd: d = yaml.load(fd) - del(d[Stage.PARAM_OUTS][0][OutputLOCAL.PARAM_MD5]) - del(d[Stage.PARAM_DEPS][0][OutputLOCAL.PARAM_MD5]) + del(d[Stage.PARAM_OUTS][0][RemoteLOCAL.PARAM_MD5]) + del(d[Stage.PARAM_DEPS][0][RemoteLOCAL.PARAM_MD5]) with open(self.file1_stage, 'w') as fd: yaml.dump(d, fd) diff --git a/tests/test_data_cloud.py b/tests/test_data_cloud.py index 8a831c3b5..a25a523b1 100644 --- a/tests/test_data_cloud.py +++ b/tests/test_data_cloud.py @@ -185,7 +185,7 @@ class TestDataCloudBase(TestDvc): # Remove and check status sleep() - shutil.rmtree(self.dvc.cache.cache_dir) + shutil.rmtree(self.dvc.cache.local.cache_dir) status = self.cloud.status(cache) self.assertEqual(status, STATUS_DELETED) @@ -283,7 +283,7 @@ class TestDataCloudCLIBase(TestDvc): sleep() self.main(['status', '-c'] + args) - shutil.rmtree(self.dvc.cache.cache_dir) + shutil.rmtree(self.dvc.cache.local.cache_dir) sleep() self.main(['status', '-c'] + args) diff --git a/tests/test_gc.py b/tests/test_gc.py index e45e4fa6e..53ac95f9e 100644 --- a/tests/test_gc.py +++ b/tests/test_gc.py @@ -17,11 +17,11 @@ class TestGC(TestDvc): self.dvc.add(self.FOO) self.dvc.add(self.DATA_DIR) - self.good_cache = self.dvc.cache.all() + self.good_cache = self.dvc.cache.local.all() self.bad_cache = [] for i in ['123', '234', '345']: - path = os.path.join(self.dvc.cache.cache_dir, i[0:2], i[2:]) + path = os.path.join(self.dvc.cache.local.cache_dir, i[0:2], i[2:]) self.create(path, i) self.bad_cache.append(path) @@ -34,7 +34,7 @@ class TestGC(TestDvc): self._test_gc() def _test_gc(self): - self.assertTrue(os.path.isdir(self.dvc.cache.cache_dir)) + self.assertTrue(os.path.isdir(self.dvc.cache.local.cache_dir)) for c in self.bad_cache: self.assertFalse(os.path.exists(c)) diff --git a/tests/test_repro.py b/tests/test_repro.py index b5f5f9cd8..d32fd261c 100644 --- a/tests/test_repro.py +++ b/tests/test_repro.py @@ -12,7 +12,7 @@ from dvc.main import main from dvc.command.repro import CmdRepro from dvc.project import ReproductionError from dvc.utils import file_md5 -from dvc.output.local import OutputLOCAL +from dvc.remote.local import RemoteLOCAL from dvc.stage import Stage from tests.basic_env import TestDvc @@ -165,8 +165,8 @@ class TestReproMissingMd5InStageFile(TestRepro): with open(self.file1_stage, 'r') as fd: d = yaml.load(fd) - del(d[Stage.PARAM_OUTS][0][OutputLOCAL.PARAM_MD5]) - del(d[Stage.PARAM_DEPS][0][OutputLOCAL.PARAM_MD5]) + del(d[Stage.PARAM_OUTS][0][RemoteLOCAL.PARAM_MD5]) + del(d[Stage.PARAM_DEPS][0][RemoteLOCAL.PARAM_MD5]) with open(self.file1_stage, 'w') as fd: yaml.dump(d, fd) diff --git a/tests/test_stage.py b/tests/test_stage.py index dbef3d336..aee553eb3 100644 --- a/tests/test_stage.py +++ b/tests/test_stage.py @@ -1,4 +1,5 @@ from dvc.output.local import OutputLOCAL +from dvc.remote.local import RemoteLOCAL from dvc.stage import Stage, StageFileFormatError from tests.basic_env import TestDvc @@ -51,8 +52,8 @@ class TestSchemaDepsOuts(TestSchema): Stage.validate(d) def test_list(self): - l = [{OutputLOCAL.PARAM_PATH: 'foo', OutputLOCAL.PARAM_MD5: '123'}, - {OutputLOCAL.PARAM_PATH: 'bar', OutputLOCAL.PARAM_MD5: None}, + l = [{OutputLOCAL.PARAM_PATH: 'foo', RemoteLOCAL.PARAM_MD5: '123'}, + {OutputLOCAL.PARAM_PATH: 'bar', RemoteLOCAL.PARAM_MD5: None}, {OutputLOCAL.PARAM_PATH: 'baz'}] d = {Stage.PARAM_DEPS: l} Stage.validate(d) From 484b04cb68e62a56d320fef47bbec6be59756020 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 4 Jun 2018 16:33:46 +0300 Subject: [PATCH 10/18] outs: raise exception if no cache location setup for s3/gs Signed-off-by: Ruslan Kuprieiev --- dvc/output/gs.py | 3 +++ dvc/output/s3.py | 3 +++ 2 files changed, 6 insertions(+) diff --git a/dvc/output/gs.py b/dvc/output/gs.py index f0d037824..65808386d 100644 --- a/dvc/output/gs.py +++ b/dvc/output/gs.py @@ -1,4 +1,5 @@ from dvc.dependency.gs import DependencyGS +from dvc.exceptions import DvcException class OutputGS(DependencyGS): @@ -7,6 +8,8 @@ class OutputGS(DependencyGS): def __init__(self, stage, path, info=None, cache=True): super(OutputGS, self).__init__(stage, path, info) self.use_cache = cache + if cache and self.project.cache.gs == None: + raise DvcException("No cache location setup for \'gs\' outputs.") def dumpd(self): ret = super(OutputGS, self).dumpd() diff --git a/dvc/output/s3.py b/dvc/output/s3.py index 12db2a8b7..5df547d86 100644 --- a/dvc/output/s3.py +++ b/dvc/output/s3.py @@ -1,4 +1,5 @@ from dvc.dependency.s3 import DependencyS3 +from dvc.exceptions import DvcException class OutputS3(DependencyS3): @@ -7,6 +8,8 @@ class OutputS3(DependencyS3): def __init__(self, stage, path, info=None, cache=True): super(OutputS3, self).__init__(stage, path, info) self.use_cache = cache + if cache and self.project.cache.s3 == None: + raise DvcException("No cache location setup for \'s3\' outputs.") def dumpd(self): ret = super(OutputS3, self).dumpd() From fc53971f2bc28a687bff5edf6b1078d2fab3a711 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 4 Jun 2018 16:47:18 +0300 Subject: [PATCH 11/18] dvc: temporarily only count local cache as used cache Signed-off-by: Ruslan Kuprieiev --- dvc/project.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/dvc/project.py b/dvc/project.py index 2718974a0..c6f3ae832 100644 --- a/dvc/project.py +++ b/dvc/project.py @@ -202,8 +202,12 @@ class Project(object): for stage in stages: for out in stage.outs: + if out.path_info['scheme'] != 'local': + continue + if not out.use_cache or not out.cache: continue + cache_set |= set([out.cache]) if self.cache.local.is_dir_cache(out.cache) and os.path.isfile(out.cache): dir_cache = self.cache.local.dir_cache(out.cache) From 42265339a80f2ed58357fb0edef3eb324f2a7e75 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 4 Jun 2018 17:14:57 +0300 Subject: [PATCH 12/18] remote: rename Remote -> RemoteBase Signed-off-by: Ruslan Kuprieiev --- dvc/remote/base.py | 2 +- dvc/remote/gs.py | 4 ++-- dvc/remote/local.py | 4 ++-- dvc/remote/s3.py | 4 ++-- dvc/remote/ssh.py | 4 ++-- 5 files changed, 9 insertions(+), 9 deletions(-) diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 1906cb356..ff187c847 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -1,4 +1,4 @@ -class Remote(object): +class RemoteBase(object): def __init__(self, project, config): pass diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index 74b85c4c1..1d4d4554f 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -6,10 +6,10 @@ try: except ImportError: from urllib.parse import urlparse -from dvc.remote.base import Remote +from dvc.remote.base import RemoteBase -class RemoteGS(Remote): +class RemoteGS(RemoteBase): PARAM_ETAG = 'etag' def __init__(self, project, config): diff --git a/dvc/remote/local.py b/dvc/remote/local.py index 1b3abe78e..4d187089b 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -4,14 +4,14 @@ import json import shutil from dvc.system import System -from dvc.remote.base import Remote +from dvc.remote.base import RemoteBase from dvc.state import State from dvc.logger import Logger from dvc.utils import remove, move from dvc.output.base import OutputAlreadyTrackedError -class RemoteLOCAL(Remote): +class RemoteLOCAL(RemoteBase): PARAM_MD5 = State.PARAM_MD5 CACHE_DIR = 'cache' diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 95cfe03f5..1cad13a3a 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -6,10 +6,10 @@ try: except ImportError: from urllib.parse import urlparse -from dvc.remote.base import Remote +from dvc.remote.base import RemoteBase -class RemoteS3(Remote): +class RemoteS3(RemoteBase): PARAM_ETAG = 'etag' def __init__(self, project, config): diff --git a/dvc/remote/ssh.py b/dvc/remote/ssh.py index 3e83c423e..3b5810172 100644 --- a/dvc/remote/ssh.py +++ b/dvc/remote/ssh.py @@ -1,5 +1,5 @@ -from dvc.remote.base import Remote +from dvc.remote.base import RemoteBase -class RemoteSSH(Remote): +class RemoteSSH(RemoteBase): pass From 1ef21039c48821d5b61638cc8b5963bebe49bb07 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 4 Jun 2018 17:24:24 +0300 Subject: [PATCH 13/18] remote: use Remote() func to determine remote type --- dvc/cache.py | 40 +++++++++++++++++----------------------- dvc/config.py | 1 + dvc/output/__init__.py | 4 ++-- dvc/remote/__init__.py | 13 +++++++++++++ dvc/remote/base.py | 16 ++++++++++++++++ dvc/remote/gs.py | 1 + dvc/remote/local.py | 21 +++++++-------------- dvc/remote/s3.py | 1 + dvc/remote/ssh.py | 2 ++ 9 files changed, 60 insertions(+), 39 deletions(-) diff --git a/dvc/cache.py b/dvc/cache.py index 64c4204be..ec2f36627 100644 --- a/dvc/cache.py +++ b/dvc/cache.py @@ -1,10 +1,11 @@ -from dvc.remote.local import RemoteLOCAL -from dvc.remote.s3 import RemoteS3 -from dvc.remote.gs import RemoteGS -from dvc.remote.ssh import RemoteSSH +import os + +from dvc.remote import Remote class Cache(object): + CACHE_DIR = 'cache' + def __init__(self, project): #FIXME from dvc.config import Config @@ -16,29 +17,22 @@ class Cache(object): sect = project.config._config[Config.SECTION_REMOTE_FMT.format(local)] else: sect = {} - d = config.get(Config.SECTION_CACHE_DIR, None) - if d: - sect[Config.SECTION_REMOTE_URL] = d + cache_dir = os.path.abspath(os.path.realpath(os.path.join(project.dvc_dir, self.CACHE_DIR))) + sect[Config.SECTION_REMOTE_URL] = config.get(Config.SECTION_CACHE_DIR, cache_dir) t = config.get(Config.SECTION_CACHE_TYPE, None) if t: sect[Config.SECTION_CACHE_TYPE] = t - self.local = RemoteLOCAL(project, sect) + self.local = Remote(project, sect) - self.s3 = None - s3 = config.get(Config.SECTION_CACHE_S3, None) - if s3: - sect = project.config._config[Config.SECTION_REMOTE_FMT.format(s3)] - self.s3 = RemoteS3(project, sect) + self.s3 = self._get_remote(project, config, Config.SECTION_CACHE_S3) + self.gs = self._get_remote(project, config, Config.SECTION_CACHE_GS) + self.ssh = self._get_remote(project, config, Config.SECTION_CACHE_SSH) - self.gs = None - gs = config.get(Config.SECTION_CACHE_GS, None) - if gs: - sect = project.config._config[Config.SECTION_REMOTE_FMT.format(gs)] - self.gs = RemoteGS(project, sect) + def _get_remote(self, project, config, name): + remote = config.get(name, None) + if not remote: + return None -# self.ssh = None -# ssh = config.get(Config.SECTION_CACHE_SSH, None) -# if ssh: -# sect = project.config._config[Config.SECTION_REMOTE_FMT.format(ssh)] -# self.ssh = RemoteSSH(project, sect) + sect = project.config._config[Config.SECTION_REMOTE_FMT.format(remote)] + return Remote(project, sect) diff --git a/dvc/config.py b/dvc/config.py index 7de2c5127..76777d630 100644 --- a/dvc/config.py +++ b/dvc/config.py @@ -35,6 +35,7 @@ class Config(object): SECTION_CACHE_LOCAL = 'local' SECTION_CACHE_S3 = 's3' SECTION_CACHE_GS = 'gs' + SECTION_CACHE_SSH = 'ssh' SECTION_CACHE_SCHEMA = { schema.Optional(SECTION_CACHE_LOCAL): str, schema.Optional(SECTION_CACHE_S3): str, diff --git a/dvc/output/__init__.py b/dvc/output/__init__.py index 0b402c83b..a14249b93 100644 --- a/dvc/output/__init__.py +++ b/dvc/output/__init__.py @@ -2,7 +2,7 @@ import schema from dvc.exceptions import DvcException -import dvc.dependency as dependency +from dvc.dependency import SCHEMA as DEP_SCHEMA from dvc.dependency.base import DependencyBase from dvc.output.s3 import OutputS3 from dvc.output.gs import OutputGS @@ -11,7 +11,7 @@ from dvc.output.local import OutputLOCAL OUTS = [OutputS3, OutputGS, OutputLOCAL] -SCHEMA = dependency.SCHEMA +SCHEMA = DEP_SCHEMA SCHEMA[schema.Optional(OutputLOCAL.PARAM_CACHE)] = bool diff --git a/dvc/remote/__init__.py b/dvc/remote/__init__.py index e69de29bb..10d2c43ed 100644 --- a/dvc/remote/__init__.py +++ b/dvc/remote/__init__.py @@ -0,0 +1,13 @@ +from dvc.remote.local import RemoteLOCAL +from dvc.remote.s3 import RemoteS3 +from dvc.remote.gs import RemoteGS + +from dvc.config import Config +from dvc.exceptions import DvcException + + +def Remote(project, config): + for r in [RemoteLOCAL, RemoteS3, RemoteGS]: + if r.supported(config): + return r(project, config) + raise DvcException('Remote \'{}\' is not supported.'.format(config)) diff --git a/dvc/remote/base.py b/dvc/remote/base.py index ff187c847..5cbb1bcba 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -1,7 +1,23 @@ +import re + + class RemoteBase(object): + REGEX = None + def __init__(self, project, config): pass + @classmethod + def supported(cls, config): + #FIXME + from dvc.config import Config + url = config[Config.SECTION_REMOTE_URL] + return cls.match(url) != None + + @classmethod + def match(cls, url): + return re.match(cls.REGEX, url) + def save_info(self, path_info): pass diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index 1d4d4554f..fd305c5f0 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -10,6 +10,7 @@ from dvc.remote.base import RemoteBase class RemoteGS(RemoteBase): + REGEX = r'^gs://(?P.*)$' PARAM_ETAG = 'etag' def __init__(self, project, config): diff --git a/dvc/remote/local.py b/dvc/remote/local.py index 4d187089b..af5beb591 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -8,13 +8,12 @@ from dvc.remote.base import RemoteBase from dvc.state import State from dvc.logger import Logger from dvc.utils import remove, move -from dvc.output.base import OutputAlreadyTrackedError class RemoteLOCAL(RemoteBase): + REGEX = r'^(?P(/+|.:\\+).*)$' PARAM_MD5 = State.PARAM_MD5 - CACHE_DIR = 'cache' CACHE_TYPES = ['reflink', 'hardlink', 'symlink', 'copy'] CACHE_TYPE_MAP = { 'copy': shutil.copyfile, @@ -28,23 +27,14 @@ class RemoteLOCAL(RemoteBase): from dvc.config import Config self.project = project - link_state = project.link_state - cache_dir = config.get(Config.SECTION_REMOTE_URL, self.CACHE_DIR) - cache_type = config.get(Config.SECTION_CACHE_TYPE, None) - - self.cache_type = cache_type - - cache_dir = cache_dir if cache_dir else self.CACHE_DIR - if os.path.isabs(cache_dir): - self.cache_dir = cache_dir - else: - self.cache_dir = os.path.abspath(os.path.realpath(os.path.join(project.dvc_dir, cache_dir))) + self.link_state = project.link_state + self.cache_dir = config[Config.SECTION_REMOTE_URL] + self.cache_type = config.get(Config.SECTION_CACHE_TYPE, None) if not os.path.exists(self.cache_dir): os.mkdir(self.cache_dir) self.state = State(self.cache_dir) - self.link_state = link_state def all(self): clist = [] @@ -222,6 +212,9 @@ class RemoteLOCAL(RemoteBase): return {self.PARAM_MD5: md5} def save(self, path_info): + #FIXME + from dvc.output.base import OutputAlreadyTrackedError + if path_info['scheme'] != 'local': raise NotImplementedError diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 1cad13a3a..99c15de97 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -10,6 +10,7 @@ from dvc.remote.base import RemoteBase class RemoteS3(RemoteBase): + REGEX = r'^s3://(?P.*)$' PARAM_ETAG = 'etag' def __init__(self, project, config): diff --git a/dvc/remote/ssh.py b/dvc/remote/ssh.py index 3b5810172..a01b7fd1e 100644 --- a/dvc/remote/ssh.py +++ b/dvc/remote/ssh.py @@ -2,4 +2,6 @@ from dvc.remote.base import RemoteBase class RemoteSSH(RemoteBase): + #NOTE: ~/ paths are temporarily forbidden + REGEX = r'^(?P.*)@(?P.*):(?P/+.*)$' pass From 7a7c9e77132895731229df1b0473643e48160cd3 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 4 Jun 2018 20:48:18 +0300 Subject: [PATCH 14/18] dvc: fix import conflicts Signed-off-by: Ruslan Kuprieiev --- dvc/cache.py | 4 +--- dvc/cloud/aws.py | 7 ++----- dvc/cloud/base.py | 10 ++-------- dvc/dependency/__init__.py | 9 ++++----- dvc/dependency/local.py | 6 ++---- dvc/output/local.py | 4 +++- dvc/remote/base.py | 4 ++-- dvc/remote/gs.py | 4 +--- dvc/remote/local.py | 10 +--------- dvc/remote/s3.py | 5 +---- 10 files changed, 19 insertions(+), 44 deletions(-) diff --git a/dvc/cache.py b/dvc/cache.py index ec2f36627..fd854223a 100644 --- a/dvc/cache.py +++ b/dvc/cache.py @@ -1,5 +1,6 @@ import os +from dvc.config import Config from dvc.remote import Remote @@ -7,9 +8,6 @@ class Cache(object): CACHE_DIR = 'cache' def __init__(self, project): - #FIXME - from dvc.config import Config - config = project.config._config[Config.SECTION_CACHE] local = config.get(Config.SECTION_CACHE_LOCAL, None) diff --git a/dvc/cloud/aws.py b/dvc/cloud/aws.py index 094a777bb..c8da5e129 100644 --- a/dvc/cloud/aws.py +++ b/dvc/cloud/aws.py @@ -6,7 +6,8 @@ import boto3 from dvc.logger import Logger from dvc.progress import progress -from dvc.cloud.base import DataCloudBase +from dvc.cloud.base import DataCloudBase, DataCloudError +from dvc.config import Config def sizeof_fmt(num, suffix='B'): @@ -56,13 +57,9 @@ class DataCloudAWS(DataCloudBase): @property def profile(self): - #FIXME - from dvc.config import Config return self._cloud_settings.cloud_config.get(Config.SECTION_AWS_PROFILE, None) def connect(self): - #FIXME - from dvc.cloud.base import DataCloudError session = boto3.Session(profile_name=self.profile) self.s3 = session.resource('s3') bucket = self.s3.Bucket(self.storage_bucket) diff --git a/dvc/cloud/base.py b/dvc/cloud/base.py index b44446ef6..7caabcafb 100644 --- a/dvc/cloud/base.py +++ b/dvc/cloud/base.py @@ -2,8 +2,10 @@ import os import re import tempfile +from dvc.config import Config from dvc.logger import Logger from dvc.exceptions import DvcException +from dvc.remote.local import RemoteLOCAL STATUS_UNKNOWN = 0 @@ -45,8 +47,6 @@ class DataCloudBase(object): @property def url(self): - #FIXME - from dvc.config import Config config = self._cloud_settings.cloud_config return config.get(Config.SECTION_REMOTE_URL, None) @@ -72,9 +72,6 @@ class DataCloudBase(object): Precedence: Storage, then cloud specific """ - #FIXME - from dvc.config import Config - if self._cloud_settings.global_storage_path: return self._cloud_settings.global_storage_path @@ -132,9 +129,6 @@ class DataCloudBase(object): path, local = arg ret = [path] - #FIXME - from dvc.remote.local import RemoteLOCAL - if not RemoteLOCAL.is_dir_cache(path): return ret diff --git a/dvc/dependency/__init__.py b/dvc/dependency/__init__.py index 29094aa40..b5ec54259 100644 --- a/dvc/dependency/__init__.py +++ b/dvc/dependency/__init__.py @@ -7,17 +7,16 @@ from dvc.dependency.s3 import DependencyS3 from dvc.dependency.gs import DependencyGS from dvc.dependency.local import DependencyLOCAL -#FIXME PARAM_MD5, PARAM_ETAG -#from dvc.remote.local import RemoteLOCAL -#from dvc.remote.s3 import RemoteS3 +from dvc.remote.local import RemoteLOCAL +from dvc.remote.s3 import RemoteS3 DEPS = [DependencyS3, DependencyGS, DependencyLOCAL] SCHEMA = { DependencyBase.PARAM_PATH: str, - schema.Optional('md5'): schema.Or(str, None), - schema.Optional('etag'): schema.Or(str, None), + schema.Optional(RemoteLOCAL.PARAM_MD5): schema.Or(str, None), + schema.Optional(RemoteS3.PARAM_ETAG): schema.Or(str, None), } diff --git a/dvc/dependency/local.py b/dvc/dependency/local.py index f8029dd5b..56dc4bc9f 100644 --- a/dvc/dependency/local.py +++ b/dvc/dependency/local.py @@ -7,6 +7,8 @@ from dvc.dependency.base import * from dvc.logger import Logger from dvc.utils import remove from dvc.cloud.local import DataCloudLOCAL +from dvc.config import Config +from dvc.remote.local import RemoteLOCAL class DependencyLOCAL(DependencyBase): @@ -16,10 +18,6 @@ class DependencyLOCAL(DependencyBase): IsNotFileOrDirError = DependencyIsNotFileOrDirError def __init__(self, stage, path, info=None): - #FIXME - from dvc.config import Config - from dvc.remote.local import RemoteLOCAL - self.stage = stage self.project = stage.project if not os.path.isabs(path): diff --git a/dvc/output/local.py b/dvc/output/local.py index 262e2d642..816e03217 100644 --- a/dvc/output/local.py +++ b/dvc/output/local.py @@ -22,7 +22,6 @@ class OutputLOCAL(DependencyLOCAL): @property def md5(self): - #FIXME return self.info.get(self.project.cache.local.PARAM_MD5, None) @property @@ -67,6 +66,9 @@ class OutputLOCAL(DependencyLOCAL): self.project.logger.debug(msg.format(self.rel_path)) return + if self.project.scm.is_tracked(self.path): + raise OutputAlreadyTrackedError(self.rel_path) + self.info = self.project.cache.local.save(self.path_info) def remove(self): diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 5cbb1bcba..721598c1e 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -1,5 +1,7 @@ import re +from dvc.config import Config + class RemoteBase(object): REGEX = None @@ -9,8 +11,6 @@ class RemoteBase(object): @classmethod def supported(cls, config): - #FIXME - from dvc.config import Config url = config[Config.SECTION_REMOTE_URL] return cls.match(url) != None diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index fd305c5f0..8d8e658a4 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -7,6 +7,7 @@ except ImportError: from urllib.parse import urlparse from dvc.remote.base import RemoteBase +from dvc.config import Config class RemoteGS(RemoteBase): @@ -14,9 +15,6 @@ class RemoteGS(RemoteBase): PARAM_ETAG = 'etag' def __init__(self, project, config): - #FIXME - from dvc.config import Config - self.project = project self.url = config.get[Config.SECTION_REMOTE_URL] self.projectname = config.get(Config.SECTION_GCP_PROJECTNAME, None) diff --git a/dvc/remote/local.py b/dvc/remote/local.py index af5beb591..febff9892 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -8,6 +8,7 @@ from dvc.remote.base import RemoteBase from dvc.state import State from dvc.logger import Logger from dvc.utils import remove, move +from dvc.config import Config class RemoteLOCAL(RemoteBase): @@ -23,9 +24,6 @@ class RemoteLOCAL(RemoteBase): } def __init__(self, project, config): - #FIXME - from dvc.config import Config - self.project = project self.link_state = project.link_state self.cache_dir = config[Config.SECTION_REMOTE_URL] @@ -212,17 +210,11 @@ class RemoteLOCAL(RemoteBase): return {self.PARAM_MD5: md5} def save(self, path_info): - #FIXME - from dvc.output.base import OutputAlreadyTrackedError - if path_info['scheme'] != 'local': raise NotImplementedError path = path_info['path'] - if self.project.scm.is_tracked(path): - raise OutputAlreadyTrackedError(os.path.relpath(path)) - if os.path.isdir(path): return self._save_dir(path_info) else: diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 99c15de97..2dc19bfae 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -6,6 +6,7 @@ try: except ImportError: from urllib.parse import urlparse +from dvc.config import Config from dvc.remote.base import RemoteBase @@ -14,10 +15,6 @@ class RemoteS3(RemoteBase): PARAM_ETAG = 'etag' def __init__(self, project, config): - #FIXME - from dvc.config import Config - - self.project = project self.url = config[Config.SECTION_REMOTE_URL] self.region = config.get(Config.SECTION_AWS_REGION, None) From 331e3b2bf954f32a77923c06008734564c184476 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 4 Jun 2018 22:24:01 +0300 Subject: [PATCH 15/18] cache: fix cache_dir resolving Signed-off-by: Ruslan Kuprieiev --- dvc/cache.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/dvc/cache.py b/dvc/cache.py index fd854223a..b2a03b510 100644 --- a/dvc/cache.py +++ b/dvc/cache.py @@ -15,8 +15,10 @@ class Cache(object): sect = project.config._config[Config.SECTION_REMOTE_FMT.format(local)] else: sect = {} - cache_dir = os.path.abspath(os.path.realpath(os.path.join(project.dvc_dir, self.CACHE_DIR))) - sect[Config.SECTION_REMOTE_URL] = config.get(Config.SECTION_CACHE_DIR, cache_dir) + cache_dir = config.get(Config.SECTION_CACHE_DIR, self.CACHE_DIR) + if not os.path.isabs(cache_dir): + cache_dir = os.path.abspath(os.path.realpath(os.path.join(project.dvc_dir, cache_dir))) + sect[Config.SECTION_REMOTE_URL] = cache_dir t = config.get(Config.SECTION_CACHE_TYPE, None) if t: sect[Config.SECTION_CACHE_TYPE] = t From 346ef139dd4e3afac8bb4fc7d6c6b0941c4d4944 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 4 Jun 2018 22:24:46 +0300 Subject: [PATCH 16/18] out: properly remove outputs when reproducing Signed-off-by: Ruslan Kuprieiev --- dvc/output/gs.py | 6 +----- dvc/output/local.py | 2 +- dvc/output/s3.py | 2 +- dvc/remote/gs.py | 20 +++++++++++++++----- dvc/remote/local.py | 6 ++++++ dvc/remote/s3.py | 10 ++++++++++ 6 files changed, 34 insertions(+), 12 deletions(-) diff --git a/dvc/output/gs.py b/dvc/output/gs.py index 65808386d..d30b560e4 100644 --- a/dvc/output/gs.py +++ b/dvc/output/gs.py @@ -41,8 +41,4 @@ class OutputGS(DependencyGS): self.info = self.project.cache.gs.save(self.path_info) def remove(self): - blob = self.client.bucket(self.bucket).get_blob(self.key) - if not blob: - return - - blob.delete() + self.remote.remove(self.path_info) diff --git a/dvc/output/local.py b/dvc/output/local.py index 816e03217..d7986b514 100644 --- a/dvc/output/local.py +++ b/dvc/output/local.py @@ -72,4 +72,4 @@ class OutputLOCAL(DependencyLOCAL): self.info = self.project.cache.local.save(self.path_info) def remove(self): - remove(self.path) + self.remote.remove(self.path_info) diff --git a/dvc/output/s3.py b/dvc/output/s3.py index 5df547d86..3cd0aedd4 100644 --- a/dvc/output/s3.py +++ b/dvc/output/s3.py @@ -40,4 +40,4 @@ class OutputS3(DependencyS3): self.info = self.project.cache.s3.save(self.path_info) def remove(self): - self.s3.Object(self.bucket, self.key).delete() + self.remote.remove(self.path_info) diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index 8d8e658a4..c70df703a 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -16,7 +16,7 @@ class RemoteGS(RemoteBase): def __init__(self, project, config): self.project = project - self.url = config.get[Config.SECTION_REMOTE_URL] + self.url = config[Config.SECTION_REMOTE_URL] self.projectname = config.get(Config.SECTION_GCP_PROJECTNAME, None) @property @@ -55,15 +55,15 @@ class RemoteGS(RemoteBase): if not blob: raise DvcException('{} doesn\'t exist in the cloud'.format(path_info['key'])) - self.gs.bucket(self.bucket).copy_blob(blob, self.bucket, name=dest_key) + self.gs.bucket(self.bucket).copy_blob(blob, self.gs.bucket(path_info['bucket']), new_name=dest_key) - return {PARAM_ETAG: etag} + return {self.PARAM_ETAG: etag} def checkout(self, path_info, checksum_info): if path_info['scheme'] != 'gs': raise NotImplementedError - etag = checksum_info.get(PARAM_ETAG, None) + etag = checksum_info.get(self.PARAM_ETAG, None) if not etag: return @@ -72,4 +72,14 @@ class RemoteGS(RemoteBase): if not blob: raise DvcException('{} doesn\'t exist in the cloud'.format(key)) - self.gs.bucket(path_info['bucket']).copy_blob(blob, path_info['bucket'], name=path_info['key']) + self.gs.bucket(path_info['bucket']).copy_blob(blob, self.gs.bucket(self.bucket), new_name=path_info['key']) + + def remove(self, path_info): + if path_info['scheme'] != 'gs': + raise NotImplementedError + + blob = self.gs.bucket(path_info['key']).get_blob(path_info['key']) + if not blob: + return + + blob.delete() diff --git a/dvc/remote/local.py b/dvc/remote/local.py index febff9892..75602ae0a 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -225,3 +225,9 @@ class RemoteLOCAL(RemoteBase): raise NotImplementedError return {self.PARAM_MD5: self.state.update(path_info['path'])} + + def remove(self, path_info): + if path_info['scheme'] != 'local': + raise NotImplementedError + + remove(path_info['path']) diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 2dc19bfae..cfc9cb2ac 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -73,3 +73,13 @@ class RemoteS3(RemoteBase): 'Key': key} self.s3.Bucket(path_info['bucket']).copy(source, path_info['key']) + + def remove(self, path_info): + if path_info['scheme'] != 's3': + raise NotImplementedError + + try: + obj = self.s3.Object(path_info['bucket'], path_info['key']).get() + obj.delete() + except Exception: + pass From 585c27f3543ef10ca5d74d9cf330e2cf32d4c2f8 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 4 Jun 2018 22:25:15 +0300 Subject: [PATCH 17/18] cloud: use project.cache.local instead of project.cache Signed-off-by: Ruslan Kuprieiev --- dvc/cloud/aws.py | 4 ++-- dvc/cloud/gcp.py | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/dvc/cloud/aws.py b/dvc/cloud/aws.py index c8da5e129..05de62ad8 100644 --- a/dvc/cloud/aws.py +++ b/dvc/cloud/aws.py @@ -81,7 +81,7 @@ class DataCloudAWS(DataCloudBase): self._makedirs(fname) tmp_file = self.tmp_file(fname) - name = os.path.relpath(fname, self._cloud_settings.cache.cache_dir) + name = os.path.relpath(fname, self._cloud_settings.cache.local.cache_dir) if self._cmp_checksum(key, fname): Logger.debug('File "{}" matches with "{}".'.format(fname, key.name)) @@ -126,7 +126,7 @@ class DataCloudAWS(DataCloudBase): def _push_key(self, key, path): """ push, aws version """ - name = os.path.relpath(path, self._cloud_settings.cache.cache_dir) + name = os.path.relpath(path, self._cloud_settings.cache.local.cache_dir) cb = self.create_cb_push(name, path) try: self.s3.Object(key.bucket, key.name).upload_file(path, Callback=cb) diff --git a/dvc/cloud/gcp.py b/dvc/cloud/gcp.py index 563e6f720..caa58070b 100644 --- a/dvc/cloud/gcp.py +++ b/dvc/cloud/gcp.py @@ -30,7 +30,7 @@ class DataCloudGCP(DataCloudBase): def _pull_key(self, key, path, no_progress_bar=False): self._makedirs(path) - name = os.path.relpath(path, self._cloud_settings.cache.cache_dir) + name = os.path.relpath(path, self._cloud_settings.cache.local.cache_dir) tmp_file = self.tmp_file(path) if self._cmp_checksum(key, path): @@ -69,7 +69,7 @@ class DataCloudGCP(DataCloudBase): def _push_key(self, key, path): """ push, gcp version """ - name = os.path.relpath(path, self._cloud_settings.cache.cache_dir) + name = os.path.relpath(path, self._cloud_settings.cache.local.cache_dir) progress.update_target(name, 0, None) From dd610c04fbd1c4da8c28912c7d9fa4cd355c84f7 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 4 Jun 2018 22:25:48 +0300 Subject: [PATCH 18/18] test: add test for s3/gs out/dep/cache Signed-off-by: Ruslan Kuprieiev --- tests/test_repro.py | 25 +++++++++++++++++-------- 1 file changed, 17 insertions(+), 8 deletions(-) diff --git a/tests/test_repro.py b/tests/test_repro.py index d32fd261c..119b58e9a 100644 --- a/tests/test_repro.py +++ b/tests/test_repro.py @@ -10,7 +10,7 @@ from google.cloud import storage as gc from dvc.main import main from dvc.command.repro import CmdRepro -from dvc.project import ReproductionError +from dvc.project import Project, ReproductionError from dvc.utils import file_md5 from dvc.remote.local import RemoteLOCAL from dvc.stage import Stage @@ -194,15 +194,24 @@ class TestReproExternalBase(TestDvc): if not self.should_test(): return + cache = self.scheme + '://' + self.bucket + '/' + str(uuid.uuid4()) + + ret = main(['config', 'cache.' + self.scheme, 'myrepo']) + self.assertEqual(ret, 0) + ret = main(['remote', 'add', 'myrepo', cache]) + self.assertEqual(ret, 0) + + self.dvc = Project('.') + foo_key = str(uuid.uuid4()) + '/' + self.FOO bar_key = str(uuid.uuid4()) + '/' + self.BAR - foo_path = self.schema + self.bucket + '/' + foo_key - bar_path = self.schema + self.bucket + '/' + bar_key + foo_path = self.scheme + '://' + self.bucket + '/' + foo_key + bar_path = self.scheme + '://' + self.bucket + '/' + bar_key self.write(self.bucket, foo_key, 'foo') - stage = self.dvc.run(outs_no_cache=[bar_path], + stage = self.dvc.run(outs=[bar_path], deps=[foo_path], cmd='{} {} {}'.format(self.cmd, foo_path, bar_path)) @@ -218,8 +227,8 @@ class TestReproExternalS3(TestReproExternalBase): return _should_test_aws() @property - def schema(self): - return 's3://' + def scheme(self): + return 's3' @property def bucket(self): @@ -242,8 +251,8 @@ class TestReproExternalGS(TestReproExternalBase): return _should_test_gcp() @property - def schema(self): - return 'gs://' + def scheme(self): + return 'gs' @property def bucket(self):