diff --git a/dvc/cache.py b/dvc/cache.py index a3c2a17f2..b2a03b510 100644 --- a/dvc/cache.py +++ b/dvc/cache.py @@ -1,210 +1,38 @@ import os -import json -import shutil -from dvc.state import State, LinkState -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 +from dvc.config import Config +from dvc.remote import Remote class Cache(object): CACHE_DIR = 'cache' - CACHE_DIR_LOCK = 'cache.lock' - CACHE_TYPES = ['reflink', 'hardlink', 'symlink', 'copy'] - CACHE_TYPE_MAP = { - 'copy': shutil.copyfile, - 'symlink': System.symlink, - 'hardlink': System.hardlink, - 'reflink': System.reflink, - } - def __init__(self, root_dir, dvc_dir, cache_dir=None, cache_type=None): - self.cache_type = cache_type + def __init__(self, project): + 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 = {} + 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 - if not os.path.exists(self.cache_dir): - os.mkdir(self.cache_dir) + self.local = Remote(project, sect) - self.state = State(self.cache_dir) - self.link_state = LinkState(root_dir, dvc_dir) - self.lock = Lock(self.cache_dir, name=self.CACHE_DIR_LOCK) + 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) - @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 - - for cache in os.listdir(subdir): - path = os.path.join(subdir, cache) - clist.append(path) - - return clist - - def get(self, md5): - if not md5: + def _get_remote(self, project, config, name): + remote = config.get(name, None) + if not remote: 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 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): - 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 checkout(self, path, md5): - with self.lock: - return self._checkout(path, md5) - - def _save_file(self, path): - md5 = self.state.update(path) - cache = self.get(md5) - if self._changed(md5): - 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): - with self.lock: - if os.path.isdir(path): - self._save_dir(path) - else: - self._save_file(path) + sect = project.config._config[Config.SECTION_REMOTE_FMT.format(remote)] + return Remote(project, sect) diff --git a/dvc/cloud/aws.py b/dvc/cloud/aws.py index 8dd1b2d4a..05de62ad8 100644 --- a/dvc/cloud/aws.py +++ b/dvc/cloud/aws.py @@ -4,10 +4,10 @@ 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, DataCloudError +from dvc.config import Config def sizeof_fmt(num, suffix='B'): @@ -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/base.py b/dvc/cloud/base.py index 21346e654..7caabcafb 100644 --- a/dvc/cloud/base.py +++ b/dvc/cloud/base.py @@ -2,10 +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.config import Config, ConfigError -from dvc.cache import Cache +from dvc.remote.local import RemoteLOCAL STATUS_UNKNOWN = 0 @@ -72,7 +72,6 @@ class DataCloudBase(object): Precedence: Storage, then cloud specific """ - if self._cloud_settings.global_storage_path: return self._cloud_settings.global_storage_path @@ -108,7 +107,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 +129,7 @@ class DataCloudBase(object): path, local = arg ret = [path] - if not Cache.is_dir_cache(path): + if not RemoteLOCAL.is_dir_cache(path): return ret if local: @@ -146,15 +145,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/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) 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..76777d630 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,18 @@ 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_SSH = 'ssh' 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 +99,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 131ee2dc4..b5ec54259 100644 --- a/dvc/dependency/__init__.py +++ b/dvc/dependency/__init__.py @@ -7,13 +7,16 @@ from dvc.dependency.s3 import DependencyS3 from dvc.dependency.gs import DependencyGS from dvc.dependency.local import DependencyLOCAL +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(RemoteLOCAL.PARAM_MD5): schema.Or(str, None), + schema.Optional(RemoteS3.PARAM_ETAG): schema.Or(str, None), } @@ -27,13 +30,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).loadd(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).loads(stage, s)) + ret.append(_get(s)(stage, s, {})) return ret diff --git a/dvc/dependency/base.py b/dvc/dependency/base.py index 15ca934cf..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,20 +54,7 @@ class DependencyBase(object): return {} def save(self): - raise NotImplemented + raise NotImplementedError def dumpd(self): return {self.PARAM_PATH: self.path} - - @classmethod - def loadd(cls, stage, d): - path = d[cls.PARAM_PATH] - return cls(stage, path) - - @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/gs.py b/dvc/dependency/gs.py index 448d6ef8d..ebe398486 100644 --- a/dvc/dependency/gs.py +++ b/dvc/dependency/gs.py @@ -1,23 +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 - def get_etag(self): - o = urlparse(self.path) - bucket_name = o.netloc - key = o.path.lstrip('/') - - client = gc.Client() - bucket = client.bucket(bucket_name) - blob = bucket.get_blob(key) - 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 b168bf5c3..56dc4bc9f 100644 --- a/dvc/dependency/local.py +++ b/dvc/dependency/local.py @@ -7,52 +7,40 @@ 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): REGEX = r'^(?P(/+|.:\\+)?.*)$' - PARAM_PATH = 'path' - PARAM_MD5 = 'md5' - MD5_DIR_SUFFIX = '.dir' - DoesNotExistError = DependencyDoesNotExistError IsNotFileOrDirError = DependencyIsNotFileOrDirError - SCHEMA = { - PARAM_PATH: str, - schema.Optional(PARAM_MD5): schema.Or(str, None), - } - - def __init__(self, stage, path, md5=None): + def __init__(self, stage, path, info=None): self.stage = stage self.project = stage.project - self.path = os.path.abspath(os.path.normpath(path)) - self.md5 = md5 + if not os.path.isabs(path): + path = self.unixpath(path) + path = os.path.join(stage.cwd, path) + self.path = os.path.normpath(path) + 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): @@ -65,7 +53,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): @@ -79,27 +67,6 @@ class DependencyLOCAL(DependencyBase): else: path = self.path - 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) - - @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] + info = self.info + info[self.PARAM_PATH] = path + return info diff --git a/dvc/dependency/s3.py b/dvc/dependency/s3.py index 9b8bfb971..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,47 +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 - - def get_etag(self): - o = urlparse(self.path) - bucket = o.netloc - key = o.path.lstrip('/') - - session = boto3.Session() - s3 = session.resource('s3') - obj = s3.Object(bucket, key).get() - 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} - - @classmethod - def loadd(cls, stage, d): - 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) - - @classmethod - def loads_from(cls, stage, s_list): - return [cls.loads(stage, x) for x in s_list] + ret = self.info + ret[self.PARAM_PATH] = self.path + return ret diff --git a/dvc/output/__init__.py b/dvc/output/__init__.py index 23dd3b542..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 @@ -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).loadd(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).loads(stage, s, use_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 0cfa59b9c..d30b560e4 100644 --- a/dvc/output/gs.py +++ b/dvc/output/gs.py @@ -1,5 +1,44 @@ from dvc.dependency.gs import DependencyGS +from dvc.exceptions import DvcException + class OutputGS(DependencyGS): - def __init__(self, stage, path, etag=None): - raise NotImplemented + PARAM_CACHE = 'cache' + + 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() + ret[self.PARAM_CACHE] = self.use_cache + return ret + + def changed(self): + if super(OutputGS, self).changed(): + return True + + if self.use_cache and self.info != self.project.cache.gs.save_info(self.path_info): + return True + + return False + + def checkout(self): + if not self.use_cache: + return + + self.project.cache.gs.checkout(self.path_info, self.info) + + def save(self): + super(OutputGS, self).save() + + if not self.use_cache: + return + + + self.info = self.project.cache.gs.save(self.path_info) + + def remove(self): + self.remote.remove(self.path_info) diff --git a/dvc/output/local.py b/dvc/output/local.py index a21e44ebf..d7986b514 100644 --- a/dvc/output/local.py +++ b/dvc/output/local.py @@ -13,49 +13,31 @@ from dvc.utils import remove class OutputLOCAL(DependencyLOCAL): PARAM_CACHE = 'cache' - DoesNotExistError = OutputDoesNotExistError - IsNotFileOrDirError = OutputIsNotFileOrDirError + DoesNotExistError = OutputDoesNotExistError + IsNotFileOrDirError = OutputIsNotFileOrDirError - SCHEMA = DependencyLOCAL.SCHEMA - SCHEMA[schema.Optional(PARAM_CACHE)] = bool + def __init__(self, stage, path, info=None, cache=True): + super(OutputLOCAL, self).__init__(stage, path, info) + self.use_cache = cache - def __init__(self, stage, path, md5=None, use_cache=True): - super(OutputLOCAL, self).__init__(stage, path, md5=md5) - self.use_cache = use_cache + @property + def md5(self): + return self.info.get(self.project.cache.local.PARAM_MD5, None) @property def cache(self): - if not self.use_cache: - return None - - return self.project.cache.get(self.md5) + return self.project.cache.local.get(self.md5) def dumpd(self): ret = super(OutputLOCAL, self).dumpd() 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 - - @classmethod - def loads(cls, stage, s, use_cache=True): - ret = super(OutputLOCAL, cls).loads(stage, s) - 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 - 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 @@ -64,31 +46,30 @@ 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) + self.remote.remove(self.path_info) diff --git a/dvc/output/s3.py b/dvc/output/s3.py index b0971dc66..3cd0aedd4 100644 --- a/dvc/output/s3.py +++ b/dvc/output/s3.py @@ -1,6 +1,43 @@ from dvc.dependency.s3 import DependencyS3 +from dvc.exceptions import DvcException class OutputS3(DependencyS3): - def __init__(self, stage, path, etag=None): - raise NotImplemented + PARAM_CACHE = 'cache' + + 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() + ret[self.PARAM_CACHE] = self.use_cache + return ret + + def changed(self): + if super(OutputS3, self).changed(): + return True + + if self.use_cache and self.info != self.project.cache.s3.save_info(self.path_info): + return True + + return False + + def checkout(self): + if not self.use_cache: + return + + self.project.cache.s3.checkout(self.path_info, self.info) + + def save(self): + super(OutputS3, self).save() + + if not self.use_cache: + return + + self.info = self.project.cache.s3.save(self.path_info) + + def remove(self): + self.remote.remove(self.path_info) diff --git a/dvc/project.py b/dvc/project.py index eda8851f6..c6f3ae832 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 LinkState from dvc.lock import Lock from dvc.scm import SCM, Base from dvc.cache import Cache @@ -47,10 +47,9 @@ 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.state = State(self.dvc_dir) + self.link_state = LinkState(self.root_dir, 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) @@ -83,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]) @@ -94,16 +91,14 @@ class Project(object): return proj def _ignore(self): - l = [self.cache.link_state.state_file, - self.cache.link_state._lock_file.lock_file, - self.state.state_file, - self.state._lock_file.lock_file, + l = [self.link_state.state_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) @@ -191,7 +186,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: @@ -207,18 +202,22 @@ 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.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) @@ -260,7 +259,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..10d2c43ed --- /dev/null +++ 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 new file mode 100644 index 000000000..721598c1e --- /dev/null +++ b/dvc/remote/base.py @@ -0,0 +1,28 @@ +import re + +from dvc.config import Config + + +class RemoteBase(object): + REGEX = None + + def __init__(self, project, config): + pass + + @classmethod + def supported(cls, 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 + + 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..c70df703a --- /dev/null +++ b/dvc/remote/gs.py @@ -0,0 +1,85 @@ +import posixpath +from google.cloud import storage + +try: + from urlparse import urlparse +except ImportError: + from urllib.parse import urlparse + +from dvc.remote.base import RemoteBase +from dvc.config import Config + + +class RemoteGS(RemoteBase): + REGEX = r'^gs://(?P.*)$' + PARAM_ETAG = 'etag' + + def __init__(self, project, config): + self.project = project + self.url = config[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.gs.bucket(path_info['bucket']), new_name=dest_key) + + return {self.PARAM_ETAG: etag} + + def checkout(self, path_info, checksum_info): + if path_info['scheme'] != 'gs': + raise NotImplementedError + + etag = checksum_info.get(self.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, 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 new file mode 100644 index 000000000..75602ae0a --- /dev/null +++ b/dvc/remote/local.py @@ -0,0 +1,233 @@ +import os +import uuid +import json +import shutil + +from dvc.system import System +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): + REGEX = r'^(?P(/+|.:\\+).*)$' + PARAM_MD5 = State.PARAM_MD5 + + 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): + self.project = project + 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) + + 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 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'])} + + 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 new file mode 100644 index 000000000..cfc9cb2ac --- /dev/null +++ b/dvc/remote/s3.py @@ -0,0 +1,85 @@ +import boto3 +import posixpath + +try: + from urlparse import urlparse +except ImportError: + from urllib.parse import urlparse + +from dvc.config import Config +from dvc.remote.base import RemoteBase + + +class RemoteS3(RemoteBase): + REGEX = r'^s3://(?P.*)$' + PARAM_ETAG = 'etag' + + def __init__(self, project, 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']) + + 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 diff --git a/dvc/remote/ssh.py b/dvc/remote/ssh.py new file mode 100644 index 000000000..a01b7fd1e --- /dev/null +++ b/dvc/remote/ssh.py @@ -0,0 +1,7 @@ +from dvc.remote.base import RemoteBase + + +class RemoteSSH(RemoteBase): + #NOTE: ~/ paths are temporarily forbidden + REGEX = r'^(?P.*)@(?P.*):(?P/+.*)$' + 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/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 5c13660e0..119b58e9a 100644 --- a/tests/test_repro.py +++ b/tests/test_repro.py @@ -10,9 +10,9 @@ 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.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) @@ -186,56 +186,83 @@ 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 + cache = self.scheme + '://' + self.bucket + '/' + str(uuid.uuid4()) - s3 = boto3.resource('s3') + ret = main(['config', 'cache.' + self.scheme, 'myrepo']) + self.assertEqual(ret, 0) + ret = main(['remote', 'add', 'myrepo', cache]) + self.assertEqual(ret, 0) - with open(self.FOO, 'rb') as fd: - s3.Bucket(TEST_AWS_REPO_BUCKET).put_object(Key=key, Body=fd.read()) + self.dvc = Project('.') - stage = self.dvc.run(outs=[self.FOO], - deps=[path], - cmd='aws s3 cp {} {}'.format(path, self.FOO)) + foo_key = str(uuid.uuid4()) + '/' + self.FOO + bar_key = str(uuid.uuid4()) + '/' + self.BAR - with open(self.BAR, 'rb') as fd: - s3.Bucket(TEST_AWS_REPO_BUCKET).put_object(Key=key, Body=fd.read()) + 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=[bar_path], + deps=[foo_path], + cmd='{} {} {}'.format(self.cmd, foo_path, bar_path)) + + 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 scheme(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 scheme(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) 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)