|
1 | | -"""Manages source control systems (e.g. Git).""" |
2 | | -from contextlib import contextmanager |
3 | | -from typing import TYPE_CHECKING, Iterator |
4 | | - |
5 | | -from dvc.exceptions import DvcException |
6 | | -from dvc.progress import Tqdm |
7 | | -from dvc.scm.base import Base # noqa: F401 |
8 | | -from dvc.scm.git import Git |
9 | | -from dvc.scm.noscm import NoSCM |
10 | | - |
11 | | -if TYPE_CHECKING: |
12 | | - from dvc.scm.progress import GitProgressEvent |
13 | | - |
14 | | - |
15 | | -class SCMError(DvcException): |
16 | | - """Base class for source control management errors.""" |
17 | | - |
18 | | - |
19 | | -class CloneError(SCMError): |
20 | | - pass |
21 | | - |
22 | | - |
23 | | -class RevError(SCMError): |
24 | | - pass |
25 | | - |
26 | | - |
27 | | -class NoSCMError(SCMError): |
28 | | - def __init__(self): |
29 | | - msg = ( |
30 | | - "Only supported for Git repositories. If you're " |
31 | | - "seeing this error in a Git repo, try updating the DVC " |
32 | | - "configuration with `dvc config core.no_scm false`." |
33 | | - ) |
34 | | - super().__init__(msg) |
35 | | - |
36 | | - |
37 | | -class InvalidRemoteSCMRepo(SCMError): |
38 | | - pass |
39 | | - |
40 | | - |
41 | | -class GitAuthError(SCMError): |
42 | | - def __init__(self, reason: str) -> None: |
43 | | - doc = "See https://dvc.org/doc//user-guide/troubleshooting#git-auth" |
44 | | - super().__init__(f"{reason}\n{doc}") |
45 | | - |
46 | | - |
47 | | -@contextmanager |
48 | | -def map_scm_exception(with_cause: bool = False) -> Iterator[None]: |
49 | | - from dvc.scm.exceptions import SCMError as InternalSCMError |
50 | | - |
51 | | - try: |
52 | | - yield |
53 | | - except InternalSCMError as exc: |
54 | | - into = SCMError(str(exc)) |
55 | | - if with_cause: |
56 | | - raise into from exc |
57 | | - raise into |
58 | | - |
59 | | - |
60 | | -def SCM( |
61 | | - root_dir, search_parent_directories=True, no_scm=False |
62 | | -): # pylint: disable=invalid-name |
63 | | - """Returns SCM instance that corresponds to a repo at the specified |
64 | | - path. |
65 | | -
|
66 | | - Args: |
67 | | - root_dir (str): path to a root directory of the repo. |
68 | | - search_parent_directories (bool): whether to look for repo root in |
69 | | - parent directories. |
70 | | - no_scm (bool): return NoSCM if True. |
71 | | -
|
72 | | - Returns: |
73 | | - dvc.scm.base.Base: SCM instance. |
74 | | - """ |
75 | | - with map_scm_exception(): |
76 | | - if no_scm: |
77 | | - return NoSCM(root_dir, _raise_not_implemented_as=NoSCMError) |
78 | | - return Git( |
79 | | - root_dir, search_parent_directories=search_parent_directories |
80 | | - ) |
81 | | - |
82 | | - |
83 | | -class TqdmGit(Tqdm): |
84 | | - def __init__(self, *args, **kwargs): |
85 | | - kwargs.setdefault("unit", "obj") |
86 | | - super().__init__(*args, **kwargs) |
87 | | - |
88 | | - def update_git(self, event: "GitProgressEvent") -> None: |
89 | | - phase, completed, total, message, *_ = event |
90 | | - if phase: |
91 | | - message = (phase + " | " + message) if message else phase |
92 | | - if message: |
93 | | - self.postfix["info"] = f" {message} | " |
94 | | - if completed: |
95 | | - self.update_to(completed, total) |
96 | | - |
97 | | - |
98 | | -def clone(url: str, to_path: str, **kwargs): |
99 | | - from dvc.scm.exceptions import CloneError as InternalCloneError |
100 | | - |
101 | | - with TqdmGit(desc="Cloning") as pbar: |
102 | | - try: |
103 | | - return Git.clone(url, to_path, progress=pbar.update_git, **kwargs) |
104 | | - except InternalCloneError as exc: |
105 | | - raise CloneError(str(exc)) |
106 | | - |
107 | | - |
108 | | -def resolve_rev(scm: "Git", rev: str) -> str: |
109 | | - from dvc.scm.exceptions import RevError as InternalRevError |
110 | | - |
111 | | - try: |
112 | | - return scm.resolve_rev(rev) |
113 | | - except InternalRevError as exc: |
114 | | - # `scm` will only resolve git branch and tag names, |
115 | | - # if rev is not a sha it may be an abbreviated experiment name |
116 | | - if not scm.is_sha(rev) and not rev.startswith("refs/"): |
117 | | - from dvc.repo.experiments.utils import exp_refs_by_name |
118 | | - |
119 | | - ref_infos = list(exp_refs_by_name(scm, rev)) |
120 | | - if len(ref_infos) == 1: |
121 | | - return scm.get_ref(str(ref_infos[0])) |
122 | | - if len(ref_infos) > 1: |
123 | | - raise RevError(f"ambiguous Git revision '{rev}'") |
124 | | - raise RevError(str(exc)) |
0 commit comments