mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-02 07:18:25 +01:00
stage: moving things around, refactor (#3793)
* stage: extract unnecessary methods out of Stage class * stage: change naming of functions * stage: extract dump equal to utils * stage: extract computation of md5 to utils * stage: extract command run related logic in run.py * stage: reorganize run() * stage: boolean cosmetic changes * stage: reorder save() and extract some into func * stage: refactor status() into smaller functions * extract resolve_wdir into utils * merge _changed() and changed() * cosmetic changes * extract dump to utils * extract import updates/sync to imports.py * remove meta from PipelineStage * fix imports, params load * make run_stage a function, fix mocks in tests * use original cmd * rollback imports on decorators * fix ds and cc
This commit is contained in:
Saugat Pachhai
+2
-1
@@ -6,6 +6,7 @@ from typing import TYPE_CHECKING, List
|
||||
from funcy import lsplit, rpartial
|
||||
|
||||
from dvc.dependency import ParamsDependency
|
||||
from dvc.stage.utils import resolve_wdir
|
||||
from dvc.utils.collections import apply_diff
|
||||
from dvc.utils.stage import parse_stage_for_update
|
||||
|
||||
@@ -77,7 +78,7 @@ def to_pipeline_file(stage: "PipelineStage"):
|
||||
|
||||
res = [
|
||||
(stage.PARAM_CMD, stage.cmd),
|
||||
(stage.PARAM_WDIR, stage.resolve_wdir()),
|
||||
(stage.PARAM_WDIR, resolve_wdir(stage.wdir, stage.path)),
|
||||
(stage.PARAM_DEPS, sorted([d.def_path for d in deps])),
|
||||
(stage.PARAM_PARAMS, serialized_params),
|
||||
*_get_outs(stage),
|
||||
|
||||
+105
-367
@@ -1,33 +1,30 @@
|
||||
import logging
|
||||
import os
|
||||
import pathlib
|
||||
import signal
|
||||
import string
|
||||
import subprocess
|
||||
import threading
|
||||
from itertools import chain, product
|
||||
|
||||
from funcy import project
|
||||
|
||||
import dvc.dependency as dependency
|
||||
import dvc.output as output
|
||||
import dvc.prompt as prompt
|
||||
from dvc.exceptions import CheckoutError, DvcException
|
||||
from dvc.utils import dict_md5, fix_env, relpath
|
||||
from dvc.utils.fs import path_isin
|
||||
from dvc.utils import relpath
|
||||
|
||||
from . import params
|
||||
from .decorators import rwlocked, unlocked_repo
|
||||
from .exceptions import (
|
||||
MissingDataSource,
|
||||
StageCmdFailedError,
|
||||
StageCommitError,
|
||||
StagePathNotDirectoryError,
|
||||
StagePathNotFoundError,
|
||||
StagePathOutsideError,
|
||||
StageUpdateError,
|
||||
from .decorators import rwlocked
|
||||
from .exceptions import StageCommitError, StageUpdateError
|
||||
from .imports import sync_import, update_import
|
||||
from .run import run_stage
|
||||
from .utils import (
|
||||
check_circular_dependency,
|
||||
check_duplicated_arguments,
|
||||
check_missing_outputs,
|
||||
check_stage_path,
|
||||
compute_md5,
|
||||
fill_stage_dependencies,
|
||||
fill_stage_outputs,
|
||||
get_dump,
|
||||
stage_dump_eq,
|
||||
)
|
||||
from .params import OutputParams
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
# Disallow all punctuation characters except hyphen and underscore
|
||||
@@ -59,14 +56,16 @@ def create_stage(cls, repo, path, **kwargs):
|
||||
wdir = os.path.abspath(kwargs.get("wdir", None) or os.curdir)
|
||||
path = os.path.abspath(path)
|
||||
check_dvc_filename(path)
|
||||
cls._check_stage_path(repo, wdir, is_wdir=kwargs.get("wdir"))
|
||||
cls._check_stage_path(repo, os.path.dirname(path))
|
||||
check_stage_path(repo, wdir, is_wdir=kwargs.get("wdir"))
|
||||
check_stage_path(repo, os.path.dirname(path))
|
||||
|
||||
stage = loads_from(cls, repo, path, wdir, kwargs)
|
||||
stage._fill_stage_outputs(**kwargs)
|
||||
stage._fill_stage_dependencies(**kwargs)
|
||||
stage._check_circular_dependency()
|
||||
stage._check_duplicated_arguments()
|
||||
fill_stage_outputs(stage, **kwargs)
|
||||
fill_stage_dependencies(
|
||||
stage, **project(kwargs, ["deps", "erepo", "params"])
|
||||
)
|
||||
check_circular_dependency(stage)
|
||||
check_duplicated_arguments(stage)
|
||||
|
||||
if stage and stage.dvcfile.exists():
|
||||
has_persist_outs = any(out.persist for out in stage.outs)
|
||||
@@ -204,7 +203,7 @@ class Stage(params.StageParams):
|
||||
|
||||
return isinstance(self.deps[0], dependency.RepoDependency)
|
||||
|
||||
def _changed_deps(self):
|
||||
def changed_deps(self):
|
||||
if self.locked:
|
||||
return False
|
||||
|
||||
@@ -219,6 +218,9 @@ class Stage(params.StageParams):
|
||||
if self.always_changed:
|
||||
return True
|
||||
|
||||
return self._changed_deps()
|
||||
|
||||
def _changed_deps(self):
|
||||
for dep in self.deps:
|
||||
status = dep.status()
|
||||
if status:
|
||||
@@ -229,10 +231,9 @@ class Stage(params.StageParams):
|
||||
)
|
||||
)
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def _changed_outs(self):
|
||||
def changed_outs(self):
|
||||
for out in self.outs:
|
||||
status = out.status()
|
||||
if status:
|
||||
@@ -246,28 +247,26 @@ class Stage(params.StageParams):
|
||||
|
||||
return False
|
||||
|
||||
def stage_changed(self, warn=False):
|
||||
changed = self.md5 != self._compute_md5()
|
||||
def changed_stage(self, warn=False):
|
||||
changed = self.md5 != self.compute_md5()
|
||||
if changed and warn:
|
||||
logger.debug("DVC-file '{}' changed.".format(self.relpath))
|
||||
return changed
|
||||
|
||||
@rwlocked(read=["deps", "outs"])
|
||||
def changed(self):
|
||||
if self._changed():
|
||||
logger.debug("{} changed.".format(self))
|
||||
return True
|
||||
|
||||
logger.debug("{} didn't change.".format(self))
|
||||
return False
|
||||
|
||||
def _changed(self):
|
||||
# Short-circuit order: stage md5 is fast, deps are expected to change
|
||||
return (
|
||||
self.stage_changed(warn=True)
|
||||
or self._changed_deps()
|
||||
or self._changed_outs()
|
||||
is_changed = (
|
||||
# Short-circuit order: stage md5 is fast,
|
||||
# deps are expected to change
|
||||
self.changed_stage(warn=True)
|
||||
or self.changed_deps()
|
||||
or self.changed_outs()
|
||||
)
|
||||
if is_changed:
|
||||
logger.info("%s changed.", self)
|
||||
else:
|
||||
logger.info("%s didn't change.", self)
|
||||
return is_changed
|
||||
|
||||
@rwlocked(write=["outs"])
|
||||
def remove_outs(self, ignore_remove=False, force=False):
|
||||
@@ -275,13 +274,10 @@ class Stage(params.StageParams):
|
||||
for out in self.outs:
|
||||
if out.persist and not force:
|
||||
out.unprotect()
|
||||
else:
|
||||
logger.debug(
|
||||
"Removing output '{out}' of {stage}.".format(
|
||||
out=out, stage=self
|
||||
)
|
||||
)
|
||||
out.remove(ignore_remove=ignore_remove)
|
||||
continue
|
||||
|
||||
logger.debug("Removing output '{}' of {}.".format(out, self))
|
||||
out.remove(ignore_remove=ignore_remove)
|
||||
|
||||
def unprotect_outs(self):
|
||||
for out in self.outs:
|
||||
@@ -298,7 +294,7 @@ class Stage(params.StageParams):
|
||||
@rwlocked(read=["deps"], write=["outs"])
|
||||
def reproduce(self, interactive=False, **kwargs):
|
||||
|
||||
if not kwargs.get("force", False) and not self.changed():
|
||||
if not (kwargs.get("force", False) or self.changed()):
|
||||
return None
|
||||
|
||||
msg = (
|
||||
@@ -316,40 +312,9 @@ class Stage(params.StageParams):
|
||||
return self
|
||||
|
||||
def update(self, rev=None):
|
||||
if not self.is_repo_import and not self.is_import:
|
||||
if not (self.is_repo_import or self.is_import):
|
||||
raise StageUpdateError(self.relpath)
|
||||
|
||||
self.deps[0].update(rev=rev)
|
||||
locked = self.locked
|
||||
self.locked = False
|
||||
try:
|
||||
self.reproduce()
|
||||
finally:
|
||||
self.locked = locked
|
||||
|
||||
@staticmethod
|
||||
def _check_stage_path(repo, path, is_wdir=False):
|
||||
assert repo is not None
|
||||
|
||||
error_msg = "{wdir_or_path} '{path}' {{}}".format(
|
||||
wdir_or_path="stage working dir" if is_wdir else "file path",
|
||||
path=path,
|
||||
)
|
||||
|
||||
real_path = os.path.realpath(path)
|
||||
if not os.path.exists(real_path):
|
||||
raise StagePathNotFoundError(error_msg.format("does not exist"))
|
||||
|
||||
if not os.path.isdir(real_path):
|
||||
raise StagePathNotDirectoryError(
|
||||
error_msg.format("is not directory")
|
||||
)
|
||||
|
||||
proj_dir = os.path.realpath(repo.root_dir)
|
||||
if real_path != proj_dir and not path_isin(real_path, proj_dir):
|
||||
raise StagePathOutsideError(
|
||||
error_msg.format("is outside of DVC repo")
|
||||
)
|
||||
update_import(self, rev=rev)
|
||||
|
||||
@property
|
||||
def can_be_skipped(self):
|
||||
@@ -362,42 +327,15 @@ class Stage(params.StageParams):
|
||||
|
||||
@property
|
||||
def is_cached(self):
|
||||
"""
|
||||
Checks if this stage has been already ran and stored
|
||||
"""
|
||||
from dvc.remote.local import LocalRemote
|
||||
from dvc.remote.s3 import S3Remote
|
||||
|
||||
"""Checks if this stage has been already ran and stored"""
|
||||
old = self.reload()
|
||||
if old._changed_outs():
|
||||
if old.changed_outs():
|
||||
return False
|
||||
|
||||
# NOTE: need to save checksums for deps in order to compare them
|
||||
# with what is written in the old stage.
|
||||
self._save_deps()
|
||||
|
||||
old_d = old.dumpd()
|
||||
new_d = self.dumpd()
|
||||
|
||||
# NOTE: need to remove checksums from old dict in order to compare
|
||||
# it to the new one, since the new one doesn't have checksums yet.
|
||||
old_d.pop(self.PARAM_MD5, None)
|
||||
new_d.pop(self.PARAM_MD5, None)
|
||||
outs = old_d.get(self.PARAM_OUTS, [])
|
||||
for out in outs:
|
||||
out.pop(LocalRemote.PARAM_CHECKSUM, None)
|
||||
out.pop(S3Remote.PARAM_CHECKSUM, None)
|
||||
|
||||
# outs and deps are lists of dicts. To check equality, we need to make
|
||||
# them independent of the order, so, we convert them to dicts.
|
||||
combination = product(
|
||||
[old_d, new_d], [self.PARAM_DEPS, self.PARAM_OUTS]
|
||||
)
|
||||
for coll, key in combination:
|
||||
if coll.get(key):
|
||||
coll[key] = {item["path"]: item for item in coll[key]}
|
||||
|
||||
if old_d != new_d:
|
||||
self.save_deps()
|
||||
if not stage_dump_eq(Stage, old.dumpd(), self.dumpd()):
|
||||
return False
|
||||
|
||||
# NOTE: committing to prevent potential data duplication. For example
|
||||
@@ -413,97 +351,29 @@ class Stage(params.StageParams):
|
||||
|
||||
return True
|
||||
|
||||
def _fill_stage_outputs(self, **kwargs):
|
||||
assert not self.outs
|
||||
|
||||
self.outs = []
|
||||
for key in (p.value for p in OutputParams):
|
||||
self.outs += output.loads_from(
|
||||
self,
|
||||
kwargs.get(key, []),
|
||||
use_cache="no_cache" not in key,
|
||||
persist="persist" in key,
|
||||
metric="metrics" in key,
|
||||
)
|
||||
|
||||
def _fill_stage_dependencies(self, **kwargs):
|
||||
assert not self.deps
|
||||
self.deps = []
|
||||
self.deps += dependency.loads_from(
|
||||
self, kwargs.get("deps", []), erepo=kwargs.get("erepo", None)
|
||||
)
|
||||
self.deps += dependency.loads_params(self, kwargs.get("params", []))
|
||||
|
||||
def _fix_outs_deps_path(self, wdir):
|
||||
for out in chain(self.outs, self.deps):
|
||||
if out.is_in_repo:
|
||||
out.def_path = relpath(out.path_info, wdir)
|
||||
|
||||
def resolve_wdir(self):
|
||||
rel_wdir = relpath(self.wdir, os.path.dirname(self.path))
|
||||
return (
|
||||
pathlib.PurePath(rel_wdir).as_posix() if rel_wdir != "." else None
|
||||
)
|
||||
|
||||
def dumpd(self):
|
||||
return {
|
||||
key: value
|
||||
for key, value in {
|
||||
Stage.PARAM_MD5: self.md5,
|
||||
Stage.PARAM_CMD: self.cmd,
|
||||
Stage.PARAM_WDIR: self.resolve_wdir(),
|
||||
Stage.PARAM_LOCKED: self.locked,
|
||||
Stage.PARAM_DEPS: [d.dumpd() for d in self.deps],
|
||||
Stage.PARAM_OUTS: [o.dumpd() for o in self.outs],
|
||||
Stage.PARAM_ALWAYS_CHANGED: self.always_changed,
|
||||
}.items()
|
||||
if value
|
||||
}
|
||||
return get_dump(self)
|
||||
|
||||
def _compute_md5(self):
|
||||
from dvc.output.base import BaseOutput
|
||||
|
||||
d = self.dumpd()
|
||||
|
||||
# Remove md5 and meta, these should not affect stage md5
|
||||
d.pop(self.PARAM_MD5, None)
|
||||
d.pop(self.PARAM_META, None)
|
||||
|
||||
# Ignore the wdir default value. In this case DVC-file w/o
|
||||
# wdir has the same md5 as a file with the default value specified.
|
||||
# It's important for backward compatibility with pipelines that
|
||||
# didn't have WDIR in their DVC-files.
|
||||
if d.get(self.PARAM_WDIR) == ".":
|
||||
del d[self.PARAM_WDIR]
|
||||
|
||||
# NOTE: excluding parameters that don't affect the state of the
|
||||
# pipeline. Not excluding `LocalOutput.PARAM_CACHE`, because if
|
||||
# it has changed, we might not have that output in our cache.
|
||||
m = dict_md5(
|
||||
d,
|
||||
exclude=[
|
||||
self.PARAM_LOCKED,
|
||||
BaseOutput.PARAM_METRIC,
|
||||
BaseOutput.PARAM_PERSIST,
|
||||
],
|
||||
)
|
||||
def compute_md5(self):
|
||||
m = compute_md5(self)
|
||||
logger.debug("Computed {} md5: '{}'".format(self, m))
|
||||
return m
|
||||
|
||||
def _save_deps(self):
|
||||
def save(self):
|
||||
self.save_deps()
|
||||
self.save_outs()
|
||||
self.md5 = self.compute_md5()
|
||||
|
||||
self.repo.stage_cache.save(self)
|
||||
|
||||
def save_deps(self):
|
||||
for dep in self.deps:
|
||||
dep.save()
|
||||
|
||||
def save(self):
|
||||
self._save_deps()
|
||||
|
||||
def save_outs(self):
|
||||
for out in self.outs:
|
||||
out.save()
|
||||
|
||||
self.md5 = self._compute_md5()
|
||||
|
||||
self.repo.stage_cache.save(self)
|
||||
|
||||
@staticmethod
|
||||
def _changed_entries(entries):
|
||||
return [
|
||||
@@ -516,7 +386,7 @@ class Stage(params.StageParams):
|
||||
changed_deps = self._changed_entries(self.deps)
|
||||
changed_outs = self._changed_entries(self.outs)
|
||||
|
||||
if changed_deps or changed_outs or self.stage_changed():
|
||||
if changed_deps or changed_outs or self.changed_stage():
|
||||
msg = (
|
||||
"dependencies {}".format(changed_deps) if changed_deps else ""
|
||||
)
|
||||
@@ -525,7 +395,7 @@ class Stage(params.StageParams):
|
||||
msg += "md5" if not (changed_deps or changed_outs) else ""
|
||||
msg += " of {} changed. ".format(self)
|
||||
msg += "Are you sure you want to commit it?"
|
||||
if not force and not prompt.confirm(msg):
|
||||
if not (force or prompt.confirm(msg)):
|
||||
raise StageCommitError(
|
||||
"unable to commit changed {}. Use `-f|--force` to "
|
||||
"force.".format(self)
|
||||
@@ -537,162 +407,28 @@ class Stage(params.StageParams):
|
||||
for out in self.outs:
|
||||
out.commit()
|
||||
|
||||
@staticmethod
|
||||
def _warn_if_fish(executable): # pragma: no cover
|
||||
if (
|
||||
executable is None
|
||||
or os.path.basename(os.path.realpath(executable)) != "fish"
|
||||
):
|
||||
return
|
||||
|
||||
logger.warning(
|
||||
"DVC detected that you are using fish as your default "
|
||||
"shell. Be aware that it might cause problems by overwriting "
|
||||
"your current environment variables with values defined "
|
||||
"in '.fishrc', which might affect your command. See "
|
||||
"https://github.com/iterative/dvc/issues/1307. "
|
||||
)
|
||||
|
||||
def _check_circular_dependency(self):
|
||||
from dvc.exceptions import CircularDependencyError
|
||||
|
||||
circular_dependencies = set(d.path_info for d in self.deps) & set(
|
||||
o.path_info for o in self.outs
|
||||
)
|
||||
|
||||
if circular_dependencies:
|
||||
raise CircularDependencyError(str(circular_dependencies.pop()))
|
||||
|
||||
def _check_duplicated_arguments(self):
|
||||
from dvc.exceptions import ArgumentDuplicationError
|
||||
from collections import Counter
|
||||
|
||||
path_counts = Counter(edge.path_info for edge in self.deps + self.outs)
|
||||
|
||||
for path, occurrence in path_counts.items():
|
||||
if occurrence > 1:
|
||||
raise ArgumentDuplicationError(str(path))
|
||||
|
||||
@unlocked_repo
|
||||
def _run(self):
|
||||
kwargs = {"cwd": self.wdir, "env": fix_env(None), "close_fds": True}
|
||||
|
||||
if os.name == "nt":
|
||||
kwargs["shell"] = True
|
||||
cmd = self.cmd
|
||||
else:
|
||||
# NOTE: when you specify `shell=True`, `Popen` [1] will default to
|
||||
# `/bin/sh` on *nix and will add ["/bin/sh", "-c"] to your command.
|
||||
# But we actually want to run the same shell that we are running
|
||||
# from right now, which is usually determined by the `SHELL` env
|
||||
# var. So instead, we compose our command on our own, making sure
|
||||
# to include special flags to prevent shell from reading any
|
||||
# configs and modifying env, which may change the behavior or the
|
||||
# command we are running. See [2] for more info.
|
||||
#
|
||||
# [1] https://github.com/python/cpython/blob/3.7/Lib/subprocess.py
|
||||
# #L1426
|
||||
# [2] https://github.com/iterative/dvc/issues/2506
|
||||
# #issuecomment-535396799
|
||||
kwargs["shell"] = False
|
||||
executable = os.getenv("SHELL") or "/bin/sh"
|
||||
|
||||
self._warn_if_fish(executable)
|
||||
|
||||
opts = {"zsh": ["--no-rcs"], "bash": ["--noprofile", "--norc"]}
|
||||
name = os.path.basename(executable).lower()
|
||||
cmd = [executable] + opts.get(name, []) + ["-c", self.cmd]
|
||||
|
||||
main_thread = isinstance(
|
||||
threading.current_thread(), threading._MainThread
|
||||
)
|
||||
old_handler = None
|
||||
p = None
|
||||
|
||||
try:
|
||||
p = subprocess.Popen(cmd, **kwargs)
|
||||
if main_thread:
|
||||
old_handler = signal.signal(signal.SIGINT, signal.SIG_IGN)
|
||||
p.communicate()
|
||||
finally:
|
||||
if old_handler:
|
||||
signal.signal(signal.SIGINT, old_handler)
|
||||
|
||||
retcode = None if not p else p.returncode
|
||||
if retcode != 0:
|
||||
raise StageCmdFailedError(self, retcode)
|
||||
|
||||
@rwlocked(read=["deps"], write=["outs"])
|
||||
def run(self, dry=False, no_commit=False, force=False, run_cache=True):
|
||||
if (self.cmd or self.is_import) and not self.locked and not dry:
|
||||
self.remove_outs(ignore_remove=False, force=False)
|
||||
|
||||
if self.locked:
|
||||
logger.info(
|
||||
"Verifying outputs in locked {stage}".format(stage=self)
|
||||
)
|
||||
if not dry:
|
||||
self.check_missing_outputs()
|
||||
|
||||
elif self.is_import:
|
||||
logger.info(
|
||||
"Importing '{dep}' -> '{out}'".format(
|
||||
dep=self.deps[0], out=self.outs[0]
|
||||
)
|
||||
)
|
||||
if not dry:
|
||||
if (
|
||||
not force
|
||||
and not self.stage_changed(warn=True)
|
||||
and self._already_cached()
|
||||
):
|
||||
self.outs[0].checkout()
|
||||
else:
|
||||
self.deps[0].download(self.outs[0])
|
||||
elif self.is_data_source:
|
||||
msg = "Verifying data sources in {}".format(self)
|
||||
logger.info(msg)
|
||||
if not dry:
|
||||
self.check_missing_outputs()
|
||||
|
||||
if not self.locked and self.is_import:
|
||||
sync_import(self, dry, force)
|
||||
elif not self.locked and self.cmd:
|
||||
run_stage(self, dry, force, run_cache)
|
||||
else:
|
||||
args = (
|
||||
("outputs", "locked ") if self.locked else ("data sources", "")
|
||||
)
|
||||
logger.info("Verifying %s in %s%s", *args, self)
|
||||
if not dry:
|
||||
stage_cache = self.repo.stage_cache
|
||||
stage_cached = (
|
||||
not force
|
||||
and not self.is_callback
|
||||
and not self.always_changed
|
||||
and self._already_cached()
|
||||
)
|
||||
use_build_cache = False
|
||||
if not stage_cached:
|
||||
self._save_deps()
|
||||
use_build_cache = (
|
||||
not force and run_cache and stage_cache.is_cached(self)
|
||||
)
|
||||
|
||||
if use_build_cache:
|
||||
# restore stage from build cache
|
||||
self.repo.stage_cache.restore(self)
|
||||
stage_cached = self._outs_cached()
|
||||
|
||||
if stage_cached:
|
||||
logger.info("Stage is cached, skipping.")
|
||||
self.checkout()
|
||||
else:
|
||||
logger.info("Running command:\n\t{}".format(self.cmd))
|
||||
self._run()
|
||||
check_missing_outputs(self)
|
||||
|
||||
if not dry:
|
||||
self.save()
|
||||
if not no_commit:
|
||||
self.commit()
|
||||
|
||||
def check_missing_outputs(self):
|
||||
paths = [str(out) for out in self.outs if not out.exists]
|
||||
if paths:
|
||||
raise MissingDataSource(paths)
|
||||
|
||||
def _filter_outs(self, path_info):
|
||||
def _func(o):
|
||||
return path_info.isin_or_eq(o.path_info)
|
||||
@@ -726,6 +462,18 @@ class Stage(params.StageParams):
|
||||
|
||||
return checkouts
|
||||
|
||||
@rwlocked(read=["deps", "outs"])
|
||||
def status(self, check_updates=False):
|
||||
ret = []
|
||||
show_import = self.is_repo_import and check_updates
|
||||
|
||||
if not self.locked or show_import:
|
||||
self._status_deps(ret)
|
||||
self._status_outs(ret)
|
||||
self._status_always_changed(ret)
|
||||
self._status_stage(ret)
|
||||
return {self.addressing: ret} if ret else {}
|
||||
|
||||
@staticmethod
|
||||
def _status(entries):
|
||||
ret = {}
|
||||
@@ -735,40 +483,31 @@ class Stage(params.StageParams):
|
||||
|
||||
return ret
|
||||
|
||||
def stage_status(self):
|
||||
return ["changed checksum"] if self.stage_changed() else []
|
||||
|
||||
@rwlocked(read=["deps", "outs"])
|
||||
def status(self, check_updates=False):
|
||||
ret = []
|
||||
|
||||
show_import = self.is_repo_import and check_updates
|
||||
|
||||
if not self.locked or show_import:
|
||||
deps_status = self._status(self.deps)
|
||||
if deps_status:
|
||||
ret.append({"changed deps": deps_status})
|
||||
def _status_deps(self, ret):
|
||||
deps_status = self._status(self.deps)
|
||||
if deps_status:
|
||||
ret.append({"changed deps": deps_status})
|
||||
|
||||
def _status_outs(self, ret):
|
||||
outs_status = self._status(self.outs)
|
||||
if outs_status:
|
||||
ret.append({"changed outs": outs_status})
|
||||
|
||||
ret.extend(self.stage_status())
|
||||
def _status_always_changed(self, ret):
|
||||
if self.is_callback or self.always_changed:
|
||||
ret.append("always changed")
|
||||
|
||||
if ret:
|
||||
return {self.addressing: ret}
|
||||
def _status_stage(self, ret):
|
||||
if self.changed_stage():
|
||||
ret.append("changed checksum")
|
||||
|
||||
return {}
|
||||
def already_cached(self):
|
||||
return self.deps_cached() and self.outs_cached()
|
||||
|
||||
def _already_cached(self):
|
||||
return self._deps_cached() and self._outs_cached()
|
||||
|
||||
def _deps_cached(self):
|
||||
def deps_cached(self):
|
||||
return all(not dep.changed() for dep in self.deps)
|
||||
|
||||
def _outs_cached(self):
|
||||
def outs_cached(self):
|
||||
return all(
|
||||
not out.changed_cache() if out.use_cache else not out.changed()
|
||||
for out in self.outs
|
||||
@@ -791,12 +530,10 @@ class Stage(params.StageParams):
|
||||
|
||||
|
||||
class PipelineStage(Stage):
|
||||
def __init__(self, *args, name=None, meta=None, **kwargs):
|
||||
def __init__(self, *args, name=None, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self.name = name
|
||||
self.cmd_changed = False
|
||||
# This is how the Stage will discover any discrepancies
|
||||
self.meta = meta or {}
|
||||
|
||||
def __eq__(self, other):
|
||||
return super().__eq__(other) and self.name == other.name
|
||||
@@ -825,10 +562,11 @@ class PipelineStage(Stage):
|
||||
def is_cached(self):
|
||||
return self.name in self.dvcfile.stages and super().is_cached
|
||||
|
||||
def stage_status(self):
|
||||
return ["changed command"] if self.cmd_changed else []
|
||||
def _status_stage(self, ret):
|
||||
if self.cmd_changed:
|
||||
ret.append("changed command")
|
||||
|
||||
def stage_changed(self, warn=False):
|
||||
def changed_stage(self, warn=False):
|
||||
if self.cmd_changed and warn:
|
||||
logger.debug("'cmd' of {} has changed.".format(self))
|
||||
return self.cmd_changed
|
||||
|
||||
@@ -2,8 +2,8 @@ from dvc.exceptions import DvcException
|
||||
|
||||
|
||||
class StageCmdFailedError(DvcException):
|
||||
def __init__(self, stage, status=None):
|
||||
msg = "failed to run: {}".format(stage.cmd)
|
||||
def __init__(self, cmd, status=None):
|
||||
msg = "failed to run: {}".format(cmd)
|
||||
if status is not None:
|
||||
msg += ", exited with {}".format(status)
|
||||
super().__init__(msg)
|
||||
@@ -112,5 +112,5 @@ class DuplicateStageName(DvcException):
|
||||
|
||||
|
||||
class InvalidStageName(DvcException):
|
||||
def __init__(self,):
|
||||
def __init__(self):
|
||||
super().__init__("Stage name cannot contain punctuation characters.")
|
||||
|
||||
@@ -0,0 +1,33 @@
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def update_import(stage, rev=None):
|
||||
stage.deps[0].update(rev=rev)
|
||||
locked = stage.locked
|
||||
stage.locked = False
|
||||
try:
|
||||
stage.reproduce()
|
||||
finally:
|
||||
stage.locked = locked
|
||||
|
||||
|
||||
def sync_import(stage, dry=False, force=False):
|
||||
"""Synchronize import's outs to the workspace."""
|
||||
logger.info(
|
||||
"Importing '{dep}' -> '{out}'".format(
|
||||
dep=stage.deps[0], out=stage.outs[0]
|
||||
)
|
||||
)
|
||||
if dry:
|
||||
return
|
||||
|
||||
if (
|
||||
not force
|
||||
and not stage.changed_stage(warn=True)
|
||||
and stage.already_cached()
|
||||
):
|
||||
stage.outs[0].checkout()
|
||||
else:
|
||||
stage.deps[0].download(stage.outs[0])
|
||||
+4
-5
@@ -11,10 +11,10 @@ from dvc import dependency, output
|
||||
|
||||
from ..dependency import ParamsDependency
|
||||
from .exceptions import StageNameUnspecified, StageNotFound
|
||||
from .utils import fill_stage_dependencies, fill_stage_outputs
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
DEFAULT_PARAMS_FILE = ParamsDependency.DEFAULT_PARAMS_FILE
|
||||
|
||||
|
||||
@@ -126,10 +126,9 @@ class StageLoader(Mapping):
|
||||
)
|
||||
stage = loads_from(PipelineStage, dvcfile.repo, path, wdir, stage_data)
|
||||
stage.name = name
|
||||
params = stage_data.pop("params", {})
|
||||
stage._fill_stage_dependencies(**stage_data)
|
||||
stage._fill_stage_outputs(**stage_data)
|
||||
cls._load_params(stage, params)
|
||||
fill_stage_dependencies(stage, deps=stage_data.get("deps", []))
|
||||
fill_stage_outputs(stage, **stage_data)
|
||||
cls._load_params(stage, stage_data.get("params", []))
|
||||
if lock_data:
|
||||
stage.cmd_changed = lock_data.get(
|
||||
Stage.PARAM_CMD
|
||||
|
||||
@@ -0,0 +1,106 @@
|
||||
import logging
|
||||
import os
|
||||
import signal
|
||||
import subprocess
|
||||
import threading
|
||||
|
||||
from dvc.utils import fix_env
|
||||
|
||||
from .decorators import unlocked_repo
|
||||
from .exceptions import StageCmdFailedError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def warn_if_fish(executable):
|
||||
if (
|
||||
executable is None
|
||||
or os.path.basename(os.path.realpath(executable)) != "fish"
|
||||
):
|
||||
return
|
||||
|
||||
logger.warning(
|
||||
"DVC detected that you are using fish as your default "
|
||||
"shell. Be aware that it might cause problems by overwriting "
|
||||
"your current environment variables with values defined "
|
||||
"in '.fishrc', which might affect your command. See "
|
||||
"https://github.com/iterative/dvc/issues/1307. "
|
||||
)
|
||||
|
||||
|
||||
@unlocked_repo
|
||||
def cmd_run(stage, *args, **kwargs):
|
||||
kwargs = {"cwd": stage.wdir, "env": fix_env(None), "close_fds": True}
|
||||
|
||||
if os.name == "nt":
|
||||
kwargs["shell"] = True
|
||||
cmd = stage.cmd
|
||||
else:
|
||||
# NOTE: when you specify `shell=True`, `Popen` [1] will default to
|
||||
# `/bin/sh` on *nix and will add ["/bin/sh", "-c"] to your command.
|
||||
# But we actually want to run the same shell that we are running
|
||||
# from right now, which is usually determined by the `SHELL` env
|
||||
# var. So instead, we compose our command on our own, making sure
|
||||
# to include special flags to prevent shell from reading any
|
||||
# configs and modifying env, which may change the behavior or the
|
||||
# command we are running. See [2] for more info.
|
||||
#
|
||||
# [1] https://github.com/python/cpython/blob/3.7/Lib/subprocess.py
|
||||
# #L1426
|
||||
# [2] https://github.com/iterative/dvc/issues/2506
|
||||
# #issuecomment-535396799
|
||||
kwargs["shell"] = False
|
||||
executable = os.getenv("SHELL") or "/bin/sh"
|
||||
|
||||
warn_if_fish(executable)
|
||||
|
||||
opts = {"zsh": ["--no-rcs"], "bash": ["--noprofile", "--norc"]}
|
||||
name = os.path.basename(executable).lower()
|
||||
cmd = [executable] + opts.get(name, []) + ["-c", stage.cmd]
|
||||
|
||||
main_thread = isinstance(threading.current_thread(), threading._MainThread)
|
||||
old_handler = None
|
||||
p = None
|
||||
|
||||
try:
|
||||
p = subprocess.Popen(cmd, **kwargs)
|
||||
if main_thread:
|
||||
old_handler = signal.signal(signal.SIGINT, signal.SIG_IGN)
|
||||
p.communicate()
|
||||
finally:
|
||||
if old_handler:
|
||||
signal.signal(signal.SIGINT, old_handler)
|
||||
|
||||
retcode = None if not p else p.returncode
|
||||
if retcode != 0:
|
||||
raise StageCmdFailedError(stage.cmd, retcode)
|
||||
|
||||
|
||||
def run_stage(stage, dry=False, force=False, run_cache=False):
|
||||
if dry:
|
||||
return
|
||||
stage_cache = stage.repo.stage_cache
|
||||
stage_cached = (
|
||||
not force
|
||||
and not stage.is_callback
|
||||
and not stage.always_changed
|
||||
and stage.already_cached()
|
||||
)
|
||||
use_build_cache = False
|
||||
if not stage_cached:
|
||||
stage.save_deps()
|
||||
use_build_cache = (
|
||||
not force and run_cache and stage_cache.is_cached(stage)
|
||||
)
|
||||
|
||||
if use_build_cache:
|
||||
# restore stage from build cache
|
||||
stage.repo.stage_cache.restore(stage)
|
||||
stage_cached = stage.outs_cached()
|
||||
|
||||
if stage_cached:
|
||||
logger.info("Stage is cached, skipping.")
|
||||
stage.checkout()
|
||||
else:
|
||||
logger.info("Running command:\n\t{}".format(stage.cmd))
|
||||
cmd_run(stage)
|
||||
@@ -0,0 +1,153 @@
|
||||
import os
|
||||
import pathlib
|
||||
from itertools import product
|
||||
|
||||
from dvc import dependency, output
|
||||
from dvc.utils.fs import path_isin
|
||||
|
||||
from ..remote import LocalRemote, S3Remote
|
||||
from ..utils import dict_md5, relpath
|
||||
from .exceptions import (
|
||||
MissingDataSource,
|
||||
StagePathNotDirectoryError,
|
||||
StagePathNotFoundError,
|
||||
StagePathOutsideError,
|
||||
)
|
||||
from .params import OutputParams
|
||||
|
||||
|
||||
def check_stage_path(repo, path, is_wdir=False):
|
||||
assert repo is not None
|
||||
|
||||
error_msg = "{wdir_or_path} '{path}' {{}}".format(
|
||||
wdir_or_path="stage working dir" if is_wdir else "file path",
|
||||
path=path,
|
||||
)
|
||||
|
||||
real_path = os.path.realpath(path)
|
||||
if not os.path.exists(real_path):
|
||||
raise StagePathNotFoundError(error_msg.format("does not exist"))
|
||||
|
||||
if not os.path.isdir(real_path):
|
||||
raise StagePathNotDirectoryError(error_msg.format("is not directory"))
|
||||
|
||||
proj_dir = os.path.realpath(repo.root_dir)
|
||||
if real_path != proj_dir and not path_isin(real_path, proj_dir):
|
||||
raise StagePathOutsideError(error_msg.format("is outside of DVC repo"))
|
||||
|
||||
|
||||
def fill_stage_outputs(stage, **kwargs):
|
||||
assert not stage.outs
|
||||
|
||||
stage.outs = []
|
||||
for key in (p.value for p in OutputParams):
|
||||
stage.outs += output.loads_from(
|
||||
stage,
|
||||
kwargs.get(key, []),
|
||||
use_cache="no_cache" not in key,
|
||||
persist="persist" in key,
|
||||
metric="metrics" in key,
|
||||
)
|
||||
|
||||
|
||||
def fill_stage_dependencies(stage, deps=None, erepo=None, params=None):
|
||||
assert not stage.deps
|
||||
stage.deps = []
|
||||
stage.deps += dependency.loads_from(stage, deps or [], erepo=erepo)
|
||||
stage.deps += dependency.loads_params(stage, params or [])
|
||||
|
||||
|
||||
def check_circular_dependency(stage):
|
||||
from dvc.exceptions import CircularDependencyError
|
||||
|
||||
circular_dependencies = {d.path_info for d in stage.deps} & {
|
||||
o.path_info for o in stage.outs
|
||||
}
|
||||
|
||||
if circular_dependencies:
|
||||
raise CircularDependencyError(str(circular_dependencies.pop()))
|
||||
|
||||
|
||||
def check_duplicated_arguments(stage):
|
||||
from dvc.exceptions import ArgumentDuplicationError
|
||||
from collections import Counter
|
||||
|
||||
path_counts = Counter(edge.path_info for edge in stage.deps + stage.outs)
|
||||
|
||||
for path, occurrence in path_counts.items():
|
||||
if occurrence > 1:
|
||||
raise ArgumentDuplicationError(str(path))
|
||||
|
||||
|
||||
def check_missing_outputs(stage):
|
||||
paths = [str(out) for out in stage.outs if not out.exists]
|
||||
if paths:
|
||||
raise MissingDataSource(paths)
|
||||
|
||||
|
||||
def stage_dump_eq(stage_cls, old_d, new_d):
|
||||
# NOTE: need to remove checksums from old dict in order to compare
|
||||
# it to the new one, since the new one doesn't have checksums yet.
|
||||
old_d.pop(stage_cls.PARAM_MD5, None)
|
||||
new_d.pop(stage_cls.PARAM_MD5, None)
|
||||
outs = old_d.get(stage_cls.PARAM_OUTS, [])
|
||||
for out in outs:
|
||||
out.pop(LocalRemote.PARAM_CHECKSUM, None)
|
||||
out.pop(S3Remote.PARAM_CHECKSUM, None)
|
||||
|
||||
# outs and deps are lists of dicts. To check equality, we need to make
|
||||
# them independent of the order, so, we convert them to dicts.
|
||||
combination = product(
|
||||
[old_d, new_d], [stage_cls.PARAM_DEPS, stage_cls.PARAM_OUTS]
|
||||
)
|
||||
for coll, key in combination:
|
||||
if coll.get(key):
|
||||
coll[key] = {item["path"]: item for item in coll[key]}
|
||||
return old_d == new_d
|
||||
|
||||
|
||||
def compute_md5(stage):
|
||||
from dvc.output.base import BaseOutput
|
||||
|
||||
d = stage.dumpd()
|
||||
|
||||
# Remove md5 and meta, these should not affect stage md5
|
||||
d.pop(stage.PARAM_MD5, None)
|
||||
d.pop(stage.PARAM_META, None)
|
||||
|
||||
# Ignore the wdir default value. In this case DVC-file w/o
|
||||
# wdir has the same md5 as a file with the default value specified.
|
||||
# It's important for backward compatibility with pipelines that
|
||||
# didn't have WDIR in their DVC-files.
|
||||
if d.get(stage.PARAM_WDIR) == ".":
|
||||
del d[stage.PARAM_WDIR]
|
||||
|
||||
return dict_md5(
|
||||
d,
|
||||
exclude=[
|
||||
stage.PARAM_LOCKED,
|
||||
BaseOutput.PARAM_METRIC,
|
||||
BaseOutput.PARAM_PERSIST,
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
def resolve_wdir(wdir, path):
|
||||
rel_wdir = relpath(wdir, os.path.dirname(path))
|
||||
return pathlib.PurePath(rel_wdir).as_posix() if rel_wdir != "." else None
|
||||
|
||||
|
||||
def get_dump(stage):
|
||||
return {
|
||||
key: value
|
||||
for key, value in {
|
||||
stage.PARAM_MD5: stage.md5,
|
||||
stage.PARAM_CMD: stage.cmd,
|
||||
stage.PARAM_WDIR: resolve_wdir(stage.wdir, stage.path),
|
||||
stage.PARAM_LOCKED: stage.locked,
|
||||
stage.PARAM_DEPS: [d.dumpd() for d in stage.deps],
|
||||
stage.PARAM_OUTS: [o.dumpd() for o in stage.outs],
|
||||
stage.PARAM_ALWAYS_CHANGED: stage.always_changed,
|
||||
}.items()
|
||||
if value
|
||||
}
|
||||
@@ -859,7 +859,9 @@ class TestReproExternalBase(SingleStageRun, TestDvc):
|
||||
stage.outs[0], "checkout", wraps=stage.outs[0].checkout
|
||||
)
|
||||
|
||||
patch_run = patch.object(stage, "_run", wraps=stage._run)
|
||||
from dvc.stage.run import cmd_run
|
||||
|
||||
patch_run = patch("dvc.stage.run.cmd_run", wraps=cmd_run)
|
||||
|
||||
with self.dvc.lock, self.dvc.state:
|
||||
with patch_download as mock_download:
|
||||
|
||||
@@ -919,7 +919,9 @@ class TestShouldNotCheckoutUponCorruptedLocalHardlinkCache(TestDvc):
|
||||
patch_checkout = mock.patch.object(
|
||||
stage.outs[0], "checkout", wraps=stage.outs[0].checkout
|
||||
)
|
||||
patch_run = mock.patch.object(stage, "_run", wraps=stage._run)
|
||||
from dvc.stage.run import cmd_run
|
||||
|
||||
patch_run = mock.patch("dvc.stage.run.cmd_run", wraps=cmd_run)
|
||||
|
||||
with self.dvc.lock, self.dvc.state:
|
||||
with patch_checkout as mock_checkout:
|
||||
|
||||
@@ -160,7 +160,7 @@ def test_md5_ignores_comments(tmp_dir, dvc):
|
||||
f.write("# End comment\n")
|
||||
|
||||
new_stage = SingleStageFile(dvc, stage.path).stage
|
||||
assert not new_stage.stage_changed()
|
||||
assert not new_stage.changed_stage()
|
||||
|
||||
|
||||
def test_meta_is_preserved(tmp_dir, dvc):
|
||||
|
||||
@@ -23,7 +23,7 @@ def test_stage_checksum():
|
||||
stage = Stage(None, "path")
|
||||
|
||||
with mock.patch.object(stage, "dumpd", return_value=TEST_STAGE_DICT):
|
||||
assert stage._compute_md5() == "e9521a22111493406ea64a88cda63e0b"
|
||||
assert stage.compute_md5() == "e9521a22111493406ea64a88cda63e0b"
|
||||
|
||||
|
||||
def test_wdir_default_ignored():
|
||||
@@ -31,7 +31,7 @@ def test_wdir_default_ignored():
|
||||
d = dict(TEST_STAGE_DICT, wdir=".")
|
||||
|
||||
with mock.patch.object(stage, "dumpd", return_value=d):
|
||||
assert stage._compute_md5() == "e9521a22111493406ea64a88cda63e0b"
|
||||
assert stage.compute_md5() == "e9521a22111493406ea64a88cda63e0b"
|
||||
|
||||
|
||||
def test_wdir_non_default_is_not_ignored():
|
||||
@@ -39,7 +39,7 @@ def test_wdir_non_default_is_not_ignored():
|
||||
d = dict(TEST_STAGE_DICT, wdir="..")
|
||||
|
||||
with mock.patch.object(stage, "dumpd", return_value=d):
|
||||
assert stage._compute_md5() == "2ceba15e87f6848aa756502c1e6d24e9"
|
||||
assert stage.compute_md5() == "2ceba15e87f6848aa756502c1e6d24e9"
|
||||
|
||||
|
||||
def test_meta_ignored():
|
||||
@@ -47,7 +47,7 @@ def test_meta_ignored():
|
||||
d = dict(TEST_STAGE_DICT, meta={"author": "Suor"})
|
||||
|
||||
with mock.patch.object(stage, "dumpd", return_value=d):
|
||||
assert stage._compute_md5() == "e9521a22111493406ea64a88cda63e0b"
|
||||
assert stage.compute_md5() == "e9521a22111493406ea64a88cda63e0b"
|
||||
|
||||
|
||||
class TestPathConversion(TestCase):
|
||||
@@ -142,7 +142,7 @@ def test_stage_cache(tmp_dir, dvc, run_copy, mocker):
|
||||
assert os.listdir(cache_dir) == [os.path.basename(cache_file)]
|
||||
assert os.path.isfile(cache_file)
|
||||
|
||||
run_spy = mocker.spy(stage, "_run")
|
||||
run_spy = mocker.patch("dvc.stage.run.cmd_run")
|
||||
checkout_spy = mocker.spy(stage, "checkout")
|
||||
with dvc.lock, dvc.state:
|
||||
stage.run()
|
||||
|
||||
Reference in New Issue
Block a user