From cf7f125c14b6b71e5134748b5ed650466b919033 Mon Sep 17 00:00:00 2001 From: Saugat Pachhai Date: Fri, 15 May 2020 04:23:32 +0000 Subject: [PATCH] refactor: simplify checkouts and run, get rid of check_can_commit (#3800) * stage: split checkouts * stage: simplify can check commit * split shell executable expansion to separate function * refactor changed_entries for better message on pipeline file * fix commit msg entry for stage * reduce complexity of run_stage * simplify prompt for commit * fixes * refactor commit * refactor run_stage * commit refactor --- dvc/repo/checkout.py | 6 ++-- dvc/repo/commit.py | 37 ++++++++++++++++++++- dvc/stage/__init__.py | 77 +++++++++++++++++++------------------------ dvc/stage/run.py | 60 +++++++++++++++++---------------- 4 files changed, 104 insertions(+), 76 deletions(-) diff --git a/dvc/repo/checkout.py b/dvc/repo/checkout.py index 3cb29b0f1..9adda0e04 100644 --- a/dvc/repo/checkout.py +++ b/dvc/repo/checkout.py @@ -86,9 +86,9 @@ def _checkout( relink=relink, filter_info=filter_info, ) - for data in ["failed", "added", "modified"]: - stats[data].extend( - _fspath_dir(path, self.root_dir) for path in result[data] + for key, items in result.items(): + stats[key].extend( + _fspath_dir(path, self.root_dir) for path in items ) if stats.get("failed"): diff --git a/dvc/repo/commit.py b/dvc/repo/commit.py index 74c98c51c..4e6329815 100644 --- a/dvc/repo/commit.py +++ b/dvc/repo/commit.py @@ -1,13 +1,48 @@ +from dvc import prompt from dvc.dvcfile import Dvcfile +from dvc.stage.exceptions import StageCommitError from . import locked +def _prepare_message(stage, changes): + changed_deps, changed_outs, changed_stage = changes + if changed_deps and changed_outs: + msg = "dependencies {deps} and outputs {outs} of {stage} changed." + elif changed_deps: + msg = "dependencies {deps} of {stage} changed." + elif changed_outs: + msg = "outputs {outs} of {stage} changed." + else: + msg = "{stage_changed}" + + msg += " Are you sure you want to commit it?" + + kw = { + "stage": stage, + "deps": changed_deps, + "outs": changed_outs, + "stage_changed": changed_stage, + } + return msg.format_map(kw) + + +def prompt_to_commit(stage, changes, force=False): + if not (force or prompt.confirm(_prepare_message(stage, changes))): + raise StageCommitError( + "unable to commit changed {}. Use `-f|--force` to " + "force.".format(stage) + ) + + @locked def commit(self, target, with_deps=False, recursive=False, force=False): stages = self.collect(target, with_deps=with_deps, recursive=recursive) for stage in stages: - stage.check_can_commit(force=force) + changes = stage.changed_entries() + if any(changes): + prompt_to_commit(stage, changes, force=force) + stage.save() stage.commit() Dvcfile(self, stage.path).dump(stage) diff --git a/dvc/stage/__init__.py b/dvc/stage/__init__.py index 97f62a1c6..0456c7948 100644 --- a/dvc/stage/__init__.py +++ b/dvc/stage/__init__.py @@ -1,6 +1,7 @@ import logging import os import string +from collections import defaultdict from funcy import project @@ -11,7 +12,7 @@ from dvc.utils import relpath from . import params from .decorators import rwlocked -from .exceptions import StageCommitError, StageUpdateError +from .exceptions import StageUpdateError from .imports import sync_import, update_import from .run import run_stage from .utils import ( @@ -250,7 +251,7 @@ class Stage(params.StageParams): def changed_stage(self, warn=False): changed = self.md5 != self.compute_md5() if changed and warn: - logger.debug("DVC-file '{}' changed.".format(self.relpath)) + logger.debug(self._changed_stage_entry()) return changed @rwlocked(read=["deps", "outs"]) @@ -382,25 +383,17 @@ class Stage(params.StageParams): if entry.checksum and entry.changed_checksum() ] - def check_can_commit(self, force): + def _changed_stage_entry(self): + return "'md5' of {} changed.".format(self) + + def changed_entries(self): changed_deps = self._changed_entries(self.deps) changed_outs = self._changed_entries(self.outs) - - if changed_deps or changed_outs or self.changed_stage(): - msg = ( - "dependencies {}".format(changed_deps) if changed_deps else "" - ) - msg += " and " if (changed_deps and changed_outs) else "" - msg += "outputs {}".format(changed_outs) if changed_outs else "" - 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 or prompt.confirm(msg)): - raise StageCommitError( - "unable to commit changed {}. Use `-f|--force` to " - "force.".format(self) - ) - self.save() + return ( + changed_deps, + changed_outs, + self._changed_stage_entry() if self.changed_stage() else None, + ) @rwlocked(write=["outs"]) def commit(self): @@ -436,31 +429,24 @@ class Stage(params.StageParams): return filter(_func, self.outs) if path_info else self.outs @rwlocked(write=["outs"]) - def checkout( - self, - force=False, - progress_callback=None, - relink=False, - filter_info=None, - ): - checkouts = {"failed": [], "added": [], "modified": []} - for out in self._filter_outs(filter_info): - try: - result = out.checkout( - force=force, - progress_callback=progress_callback, - relink=relink, - filter_info=filter_info, - ) - added, modified = result or (None, None) - if modified: - checkouts["modified"].append(out.path_info) - elif added: - checkouts["added"].append(out.path_info) - except CheckoutError as exc: - checkouts["failed"].extend(exc.target_infos) + def checkout(self, **kwargs): + stats = defaultdict(list) + for out in self._filter_outs(kwargs.get("filter_info")): + key, outs = self._checkout(out, **kwargs) + if key: + stats[key].extend(outs) + return stats - return checkouts + @staticmethod + def _checkout(out, **kwargs): + try: + result = out.checkout(**kwargs) + added, modified = result or (None, None) + if not (added or modified): + return None, [] + return "modified" if modified else "added", [out.path_info] + except CheckoutError as exc: + return "failed", exc.target_infos @rwlocked(read=["deps", "outs"]) def status(self, check_updates=False): @@ -568,5 +554,8 @@ class PipelineStage(Stage): def changed_stage(self, warn=False): if self.cmd_changed and warn: - logger.debug("'cmd' of {} has changed.".format(self)) + logger.debug(self._changed_stage_entry()) return self.cmd_changed + + def _changed_stage_entry(self): + return "'cmd' of {} has changed.".format(self) diff --git a/dvc/stage/run.py b/dvc/stage/run.py index 4e33f96df..54b1bfbc7 100644 --- a/dvc/stage/run.py +++ b/dvc/stage/run.py @@ -12,6 +12,12 @@ from .exceptions import StageCmdFailedError logger = logging.getLogger(__name__) +def _nix_cmd(executable, cmd): + opts = {"zsh": ["--no-rcs"], "bash": ["--noprofile", "--norc"]} + name = os.path.basename(executable).lower() + return [executable] + opts.get(name, []) + ["-c", cmd] + + def warn_if_fish(executable): if ( executable is None @@ -51,12 +57,8 @@ def cmd_run(stage, *args, **kwargs): # #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] + cmd = _nix_cmd(executable, stage.cmd) main_thread = isinstance(threading.current_thread(), threading._MainThread) old_handler = None @@ -76,32 +78,34 @@ def cmd_run(stage, *args, **kwargs): raise StageCmdFailedError(stage.cmd, retcode) -def run_stage(stage, dry=False, force=False, run_cache=False): - if dry: - logger.info("Running command:\n\t{}".format(stage.cmd)) - return - stage_cache = stage.repo.stage_cache - stage_cached = ( - not force - and not stage.is_callback +def _is_cached(stage): + return ( + 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) + + +def restored_from_cache(stage): + stage.save_deps() + stage_cache = stage.repo.stage_cache + if not stage_cache.is_cached(stage): + return False + # restore stage from build cache + stage_cache.restore(stage) + return stage.outs_cached() + + +def run_stage(stage, dry=False, force=False, run_cache=False): + if not (dry or force): + stage_cached = _is_cached(stage) or ( + run_cache and restored_from_cache(stage) ) + if stage_cached: + logger.info("Stage is cached, skipping.") + stage.checkout() + return - 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)) + logger.info("Running command:\n\t%s", stage.cmd) + if not dry: cmd_run(stage)