From 421a2f473e10ac13c261f209ede54a356bc34848 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 8 Jan 2018 23:50:12 +0200 Subject: [PATCH] stage: fix defaults and RFC classes Signed-off-by: Ruslan Kuprieiev --- dvc/command/fsck.py | 6 +- dvc/project.py | 17 ++-- dvc/stage.py | 217 ++++++++++++++++++++++++-------------------- 3 files changed, 129 insertions(+), 111 deletions(-) diff --git a/dvc/command/fsck.py b/dvc/command/fsck.py index e729349ef..5695c3501 100644 --- a/dvc/command/fsck.py +++ b/dvc/command/fsck.py @@ -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) diff --git a/dvc/project.py b/dvc/project.py index c1d659ba7..8cf5f5084 100644 --- a/dvc/project.py +++ b/dvc/project.py @@ -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) diff --git a/dvc/stage.py b/dvc/stage.py index 9c32b2b7a..1039f3536 100644 --- a/dvc/stage.py +++ b/dvc/stage.py @@ -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()