stage: move is_cached logic to Stage.can_be_skipped (#4994)

Having that logic in `create_stage` is unnatural and also has some
side-effects for unsuspecting users. For example, it will check the
filesystem for existing dvcfile and will try to load a corresponding old
stage. I've adjusted it with minimal changes to better reflect the logic
behind it.

Pre-requisite for #4907

dvc/repo/add.py

@@ -181,7 +181,9 @@ def _create_stages(
             outs=[out],
             external=external,
         )
-        if stage:
+        if stage.can_be_skipped:
+            stage = None
+        else:
             Dvcfile(repo, stage.path).remove()
             if desc:
                 stage.outs[0].desc = desc

dvc/repo/imp_url.py

@@ -44,7 +44,7 @@ def imp_url(
         erepo=erepo,
     )
 
-    if stage is None:
+    if stage.can_be_skipped:
         return None
 
     if desc:

dvc/repo/run.py

@@ -113,7 +113,7 @@ def run(self, fname=None, no_exec=False, single_stage=False, **kwargs):
     stage = create_stage(
         stage_cls, repo=self, path=path, params=params, **kwargs
     )
-    if stage is None:
+    if kwargs.get("run_cache", True) and stage.can_be_skipped:
         return None
 
     dvcfile = Dvcfile(self, stage.path)

dvc/stage/__init__.py

@@ -87,18 +87,6 @@ def create_stage(cls, repo, path, external=False, **kwargs):
     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)
-        ignore_run_cache = (
-            not kwargs.get("run_cache", True) or has_persist_outs
-        )
-        if has_persist_outs:
-            logger.warning("Build cache is ignored when persisting outputs.")
-
-        if not ignore_run_cache and stage.can_be_skipped:
-            logger.info("Stage is cached, skipping")
-            return None
-
     return stage
 
 
@@ -361,9 +349,19 @@ class Stage(params.StageParams):
 
     @property
     def can_be_skipped(self):
-        return (
-            self.is_cached and not self.is_callback and not self.always_changed
-        )
+        if not self.dvcfile.exists():
+            return False
+
+        has_persist_outs = any(out.persist for out in self.outs)
+        if has_persist_outs:
+            logger.warning("Build cache is ignored when persisting outputs.")
+            return False
+
+        if self.is_cached and not self.is_callback and not self.always_changed:
+            logger.info("Stage is cached, skipping")
+            return True
+
+        return False
 
     def reload(self):
         return self.dvcfile.stage