Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-10 23:10:42 +01:00
Code Issues Packages Projects Releases Wiki Activity

dvc: use Remote concept in existing code

Browse Source 
Signed-off-by: Ruslan Kuprieiev <[email protected]>
This commit is contained in:
Ruslan Kuprieiev
2018-06-04 22:28:14 +03:00
parent 6f4d0cd74a
commit 88397ef430
31 changed files with 602 additions and 401 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/cache.py
+34 -195
View File
@@ -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)
dvc/cloud/aws.py
+5 -2
View File
@@ -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)
dvc/cloud/base.py
+13 -8
View File
@@ -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
dvc/cloud/local.py
+1 -1
View File
@@ -23,7 +23,7 @@ class DataCloudLOCAL(DataCloudBase):
REGEX = r'^(?P<path>(/+|.:\\+).*)$'
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)
dvc/cloud/ssh.py
+1 -1
View File
@@ -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):
dvc/config.py
+11 -3
View File
@@ -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 = {
dvc/dependency/__init__.py
+9 -5
View File
@@ -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
dvc/dependency/base.py
+4 -2
View File
@@ -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}
dvc/dependency/gs.py
+10 -14
View File
@@ -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('/')}
dvc/dependency/local.py
+17 -24
View File
@@ -12,45 +12,37 @@ from dvc.cloud.local import DataCloudLOCAL
class DependencyLOCAL(DependencyBase):
REGEX = r'^(?P<path>(/+|.:\\+)?.*)$'
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
dvc/dependency/s3.py
+13 -31
View File
@@ -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
dvc/output/__init__.py
+5 -4
View File
@@ -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
dvc/output/gs.py
+6 -14
View File
@@ -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)
dvc/output/local.py
+22 -24
View File
@@ -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)
dvc/output/s3.py
+5 -14
View File
@@ -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()
dvc/project.py
+9 -16
View File
@@ -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]
dvc/remote/__init__.py
View File
dvc/remote/base.py
+10 -1
View File
@@ -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
dvc/remote/gs.py
+76
View File
@@ -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'])
dvc/remote/local.py
+242
View File
@@ -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'])}
dvc/remote/s3.py
+77
View File
@@ -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'])
dvc/remote/ssh.py
+5
View File
@@ -0,0 +1,5 @@
from dvc.remote.base import Remote
class RemoteSSH(Remote):
pass
dvc/stage.py
+1 -1
View File
@@ -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):
dvc/transport/__init__.py
-21
View File
@@ -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)
tests/test_add.py
+1 -1
View File
@@ -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):
tests/test_cache.py
+11 -6
View File
@@ -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, '<clearly>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):
tests/test_checkout.py
+3 -3
View File
@@ -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)
tests/test_data_cloud.py
+2 -2
View File
@@ -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)
tests/test_gc.py
+3 -3
View File
@@ -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))
tests/test_repro.py
+3 -3
View File
@@ -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)
tests/test_stage.py
+3 -2
View File
@@ -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)