stage: fix defaults and RFC classes

Signed-off-by: Ruslan Kuprieiev <[email protected]>
This commit is contained in:
Ruslan Kuprieiev
2018-01-09 00:40:31 +02:00
parent cd12513c57
commit 421a2f473e
3 changed files with 129 additions and 111 deletions
+4 -2
View File
@@ -122,8 +122,10 @@ class CmdFsck(CmdBase):
hardlink_md5 = caches.get(dvc_path)
state = self.project.state.get(dvc_path)
fsck_deps = [FileFsckDep(stage.dvc_path, dep.md5, type(dep).__name__, dep.use_cache)
for stage, dep in files_and_stages.get(dvc_path, [])]
fsck_deps = []
for stage, dep in files_and_stages.get(dvc_path, []):
use_cache = dep.use_cache if hasattr(dep, 'use_cache') else False
fsck_deps.append(FileFsckDep(stage.dvc_path, dep.md5, type(dep).__name__, use_cache))
return FileFsck(dvc_path, full_path, md5, hardlink_md5, state, fsck_deps)
+8 -9
View File
@@ -1,5 +1,4 @@
import os
import itertools
import networkx as nx
from dvc.logger import Logger
@@ -154,7 +153,7 @@ class Project(object):
path = os.path.join(cwd, fname)
outputs = Output.loads_from(self, outs, use_cache=True, cwd=cwd)
outputs += Output.loads_from(self, outs_no_cache, use_cache=False, cwd=cwd)
deps = Dependency.loads_from(self, deps, use_cache=False, cwd=cwd)
deps = Dependency.loads_from(self, deps, cwd=cwd)
stage = Stage(project=self,
path=path,
@@ -194,11 +193,11 @@ class Project(object):
def _used_cache(self):
clist = []
for stage in self.stages():
for entry in itertools.chain(stage.outs, stage.deps):
if not entry.use_cache:
for out in stage.outs:
if not out.use_cache:
continue
if entry.cache not in clist:
clist.append(entry.cache)
if out.cache not in clist:
clist.append(out.cache)
return clist
def gc(self):
@@ -215,9 +214,9 @@ class Project(object):
def pull(self, jobs=1):
self.cloud.pull(self._used_cache(), jobs)
for stage in self.stages():
for entry in itertools.chain(stage.outs, stage.deps):
if entry.use_cache:
entry.link()
for out in stage.outs:
if out.use_cache:
out.link()
def status(self, jobs=1):
return self.cloud.status(self._used_cache(), jobs)
+117 -100
View File
@@ -39,12 +39,11 @@ class OutputAlreadyTrackedError(OutputError):
super(OutputAlreadyTrackedError, self).__init__(path, 'already tracked by scm(e.g. git)')
class Output(object):
class Dependency(object):
PARAM_PATH = 'path'
PARAM_MD5 = 'md5'
PARAM_CACHE = 'cache'
def __init__(self, project, path, md5=None, use_cache=False):
def __init__(self, project, path, md5=None):
self.project = project
self.path = os.path.abspath(os.path.realpath(path))
@@ -52,7 +51,6 @@ class Output(object):
raise OutputOutsideOfRepoError(self.path)
self.md5 = md5
self.use_cache = use_cache
@property
def dvc_path(self):
@@ -71,62 +69,7 @@ class Output(object):
return self.md5 != md5
def changed(self):
if not self.use_cache:
changed = self._changed_md5()
else:
changed = os.path.exists(self.path) \
and os.path.exists(self.cache) \
and not System.samefile(self.path, self.cache)
if changed:
self.project.logger.debug('{} changed'.format(self.path))
return changed
@property
def cache(self):
return self.project.cache.get(self.md5)
def link(self, checkout=False):
if not self.use_cache:
raise OutputNoCacheError(self.path)
if not os.path.exists(self.path) and not os.path.exists(self.cache):
raise OutputNoCacheError(self.path)
if os.path.exists(self.path) and \
os.path.exists(self.cache) and \
System.samefile(self.path, self.cache) and \
os.stat(self.cache).st_mode & stat.S_IREAD:
return
if os.path.exists(self.cache):
if os.path.exists(self.path):
# This means that we already have cache for this data.
# We remove data and link it to existing cache to save
# some space.
os.unlink(self.path)
src = self.cache
link = self.path
elif not checkout:
src = self.path
link = self.cache
else:
raise OutputNoCacheError(self.path)
System.hardlink(src, link)
os.chmod(self.path, stat.S_IREAD)
def checkout(self):
if not self.use_cache:
return
if not os.path.exists(self.cache):
self.project.logger.warn(u'\'{}\': cache file not found'.format(self.dvc_path))
if os.path.exists(self.path):
os.unlink(self.path)
else:
self.link(checkout=True)
return self._changed_md5()
def mtime(self):
return os.path.getmtime(self.path)
@@ -147,6 +90,118 @@ class Output(object):
self.md5 = file_md5(self.path)[0]
self.project.state.update(self.path, self.md5, self.mtime())
def dumpd(self, cwd):
return {
Output.PARAM_PATH: os.path.relpath(self.path, cwd),
Output.PARAM_MD5: self.md5,
}
@classmethod
def loadd(cls, project, d, cwd=os.curdir):
path = os.path.join(cwd, d[Output.PARAM_PATH])
md5 = d.get(Output.PARAM_MD5, None)
return cls(project, path, md5=md5)
@classmethod
def loadd_from(cls, project, d_list, cwd=os.curdir):
return [cls.loadd(project, x, cwd=cwd) for x in d_list]
@classmethod
def loads(cls, project, s, cwd=os.curdir):
return cls(project, os.path.join(cwd, s), md5=None)
@classmethod
def loads_from(cls, project, s_list, cwd=os.curdir):
return [cls.loads(project, x, cwd=cwd) for x in s_list]
def stage(self):
for stage in self.project.stages():
for out in stage.outs:
if self.path == out.path:
return stage
return None
class Output(Dependency):
PARAM_CACHE = 'cache'
def __init__(self, project, path, md5=None, use_cache=True):
super(Output, self).__init__(project, path, md5=md5)
self.use_cache = use_cache
@property
def cache(self):
return self.project.cache.get(self.md5)
def dumpd(self, cwd):
ret = super(Output, self).dumpd(cwd)
ret[Output.PARAM_CACHE] = self.use_cache
return ret
@classmethod
def loadd(cls, project, d, cwd=os.curdir):
ret = super(Output, cls).loadd(project, d, cwd=cwd)
ret.use_cache = d.get(Output.PARAM_CACHE, True)
return ret
@classmethod
def loads(cls, project, s, use_cache=True, cwd=os.curdir):
ret = super(Output, cls).loads(project, s, cwd=cwd)
ret.use_cache = use_cache
return ret
@classmethod
def loads_from(cls, project, s_list, use_cache=False, cwd=os.curdir):
return [cls.loads(project, x, use_cache=use_cache, cwd=cwd) for x in s_list]
def changed(self):
if not self.use_cache:
return super(Output, self).changed()
return os.path.exists(self.path) \
and os.path.exists(self.cache) \
and not System.samefile(self.path, self.cache)
def link(self, checkout=False):
if not self.use_cache:
raise OutputNoCacheError(self.path)
if not os.path.exists(self.path) and not os.path.exists(self.cache):
raise OutputNoCacheError(self.path)
if os.path.exists(self.path) and \
os.path.exists(self.cache) and \
System.samefile(self.path, self.cache) and \
os.stat(self.cache).st_mode & stat.S_IREAD:
return
if os.path.exists(self.cache):
if os.path.exists(self.path):
# This means that we already have cache for this data.
# We remove data and link it to existing cache to save
# some space.
self.remove()
src = self.cache
link = self.path
elif not checkout:
src = self.path
link = self.cache
else:
raise OutputNoCacheError(self.path)
System.hardlink(src, link)
os.chmod(self.path, stat.S_IREAD)
def checkout(self):
if not self.use_cache:
return
if not os.path.exists(self.cache):
self.project.logger.warn(u'\'{}\': cache file not found'.format(self.dvc_path))
self.remove()
else:
self.link(checkout=True)
def save(self):
if not self.use_cache:
return
@@ -157,39 +212,6 @@ class Output(object):
self.project.scm.ignore(self.path)
self.link()
def dumpd(self, cwd):
return {
Output.PARAM_PATH: os.path.relpath(self.path, cwd),
Output.PARAM_MD5: self.md5,
Output.PARAM_CACHE: self.use_cache
}
@classmethod
def loadd(cls, project, d, cwd='.'):
path = os.path.join(cwd, d[Output.PARAM_PATH])
md5 = d.get(Output.PARAM_MD5, None)
use_cache = d.get(Output.PARAM_CACHE, False)
return cls(project, path, md5, use_cache=use_cache)
@classmethod
def loadd_from(cls, project, d_list, cwd='.'):
return [cls.loadd(project, x, cwd=cwd) for x in d_list]
@classmethod
def loads(cls, project, s, use_cache=False, cwd='.'):
return cls(project, os.path.join(cwd, s), None, use_cache=use_cache)
@classmethod
def loads_from(cls, project, s_list, use_cache=False, cwd='.'):
return [cls.loads(project, x, use_cache, cwd=cwd) for x in s_list]
def stage(self):
for stage in self.project.stages():
for out in stage.outs:
if self.path == out.path:
return stage
return None
def remove(self):
if not os.path.exists(self.path):
return
@@ -201,10 +223,6 @@ class Output(object):
os.chmod(self.cache, stat.S_IREAD)
class Dependency(Output):
pass
class Stage(object):
STAGE_FILE = 'Dvcfile'
STAGE_FILE_SUFFIX = '.dvc'
@@ -311,7 +329,6 @@ class Stage(object):
def save(self):
for dep in self.deps:
dep.update()
dep.save()
for out in self.outs:
out.update()
@@ -323,5 +340,5 @@ class Stage(object):
self.save()
def checkout(self):
for entry in itertools.chain(self.outs, self.deps):
entry.checkout()
for out in self.outs:
out.checkout()