scm: map internal SCMError to DVC's exception type

dvc/external_repo.py

@@ -16,12 +16,14 @@ from dvc.exceptions import (
     PathMissingError,
 )
 from dvc.repo import Repo
+from dvc.scm import map_scm_exception
 from dvc.utils import relpath
 
 logger = logging.getLogger(__name__)
 
 
 @contextmanager
+@map_scm_exception()
 def external_repo(
     url, rev=None, for_write=False, cache_dir=None, cache_types=None, **kwargs
 ):

dvc/repo/__init__.py

@@ -286,7 +286,10 @@ class Repo:
 
         assert self.scm
         if isinstance(self.fs, LocalFileSystem):
-            return self.scm.get_rev()
+            from dvc.scm import map_scm_exception
+
+            with map_scm_exception():
+                return self.scm.get_rev()
         return self.fs.rev
 
     @cached_property

dvc/repo/experiments/__init__.py

@@ -42,7 +42,9 @@ def scm_locked(f):
     # different sequences of git operations at once
     @wraps(f)
     def wrapper(exp, *args, **kwargs):
-        with exp.scm_lock:
+        from dvc.scm import map_scm_exception
+
+        with map_scm_exception(), exp.scm_lock:
             return f(exp, *args, **kwargs)
 
     return wrapper

dvc/scm/__init__.py

@@ -1,4 +1,6 @@
 """Manages source control systems (e.g. Git)."""
+from contextlib import contextmanager
+from typing import Iterator
 
 from dvc.scm.base import Base, NoSCMError
 from dvc.scm.git import Git
@@ -11,6 +13,20 @@ class NoSCM(Base):
         raise NoSCMError
 
 
+@contextmanager
+def map_scm_exception(with_cause: bool = False) -> Iterator[None]:
+    from dvc.scm.base import SCMError
+    from dvc.scm.exceptions import SCMError as InternalSCMError
+
+    try:
+        yield
+    except InternalSCMError as exc:
+        into = SCMError(str(exc))
+        if with_cause:
+            raise into from exc
+        raise into
+
+
 def SCM(
     root_dir, search_parent_directories=True, no_scm=False
 ):  # pylint: disable=invalid-name
@@ -26,17 +42,12 @@ def SCM(
     Returns:
         dvc.scm.base.Base: SCM instance.
     """
-    from dvc.scm.base import SCMError
-    from dvc.scm.exceptions import SCMError as InternalSCMError
-
-    try:
+    with map_scm_exception():
         if no_scm:
             return NoSCM(root_dir)
         return Git(
             root_dir, search_parent_directories=search_parent_directories
         )
-    except InternalSCMError as exc:
-        raise SCMError(str(exc))
 
 
 def clone(url: str, to_path: str, **kwargs):

dvc/scm/git/__init__.py

@@ -12,11 +12,14 @@ from funcy import cached_property, first
 from pathspec.patterns import GitWildMatchPattern
 
 from dvc.scm.base import Base
-from dvc.scm.exceptions import FileNotInRepoError, RevError
+from dvc.scm.exceptions import (
+    FileNotInRepoError,
+    GitHookAlreadyExists,
+    RevError,
+)
 from dvc.utils import relpath
 from dvc.utils.fs import path_isin
 
-from ..exceptions import GitHookAlreadyExists
 from .backend.base import BaseGitBackend, NoGitBackendError
 from .backend.dulwich import DulwichBackend
 from .backend.gitpython import GitPythonBackend