index: collect_targets: introduce onerror (#8925)

dvc/repo/index.py

@@ -301,24 +301,32 @@ class Index:
         )
 
     def collect_targets(
-        self, targets: Optional["TargetType"], **kwargs: Any
+        self, targets: Optional["TargetType"], *, onerror=None, **kwargs: Any
     ) -> List["StageInfo"]:
-        from itertools import chain
-
+        from dvc.exceptions import DvcException
         from dvc.repo.stage import StageInfo
         from dvc.utils.collections import ensure_list
 
+        if not onerror:
+
+            def onerror(_target, _exc):
+                raise  # pylint: disable=misplaced-bare-raise
+
         targets = ensure_list(targets)
         if not targets:
             return [StageInfo(stage) for stage in self.stages]
         targets_hash = self._hash_targets(targets, **kwargs)
         if targets_hash not in self._collected_targets:
-            self._collected_targets[targets_hash] = list(
-                chain.from_iterable(
-                    self.repo.stage.collect_granular(target, **kwargs)
-                    for target in targets
-                )
-            )
+            collected = []
+            for target in targets:
+                try:
+                    collected.extend(
+                        self.repo.stage.collect_granular(target, **kwargs)
+                    )
+                except DvcException as exc:
+                    onerror(target, exc)
+            self._collected_targets[targets_hash] = collected
+
         return self._collected_targets[targets_hash]
 
     def used_objs(

tests/func/test_repo_index.py

@@ -212,6 +212,31 @@ def test_view_granular_dir(tmp_dir, scm, dvc, run_copy):
     ) not in data_index
 
 
+def test_view_onerror(tmp_dir, scm, dvc):
+    from dvc.exceptions import NoOutputOrStageError
+
+    tmp_dir.dvc_gen({"foo": "foo"}, commit="init")
+    index = Index.from_repo(dvc)
+
+    with pytest.raises(NoOutputOrStageError):
+        index.targets_view(["foo", "missing"])
+
+    failed = []
+
+    def onerror(target, exc):
+        failed.append((target, exc))
+
+    view = index.targets_view(["foo", "missing"], onerror=onerror)
+    data = view.data["repo"]
+
+    assert len(failed) == 1
+    target, exc = failed[0]
+    assert target == "missing"
+    assert isinstance(exc, NoOutputOrStageError)
+    assert len(data) == 1
+    assert data[("foo",)]
+
+
 def test_view_stage_filter(tmp_dir, scm, dvc, run_copy):
     (stage1,) = tmp_dir.dvc_gen("foo", "foo")
     stage2 = run_copy("foo", "bar", name="copy-foo-bar")