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

@@ -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"):

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)

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)

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)