Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import logging
4import os.path
5import pathlib
6import re
7import urllib.parse
8from dataclasses import replace
9from typing import Any
11from pip._internal.exceptions import BadCommand, InstallationError
12from pip._internal.utils.misc import HiddenText, display_path, hide_url, strtobool
13from pip._internal.utils.subprocess import make_command
14from pip._internal.vcs.versioncontrol import (
15 AuthInfo,
16 RemoteNotFoundError,
17 RemoteNotValidError,
18 RevOptions,
19 VersionControl,
20 find_path_to_project_root_from_repo_root,
21 vcs,
22)
24urlsplit = urllib.parse.urlsplit
25urlunsplit = urllib.parse.urlunsplit
28logger = logging.getLogger(__name__)
31GIT_VERSION_REGEX = re.compile(
32 r"^git version " # Prefix.
33 r"(\d+)" # Major.
34 r"\.(\d+)" # Dot, minor.
35 r"(?:\.(\d+))?" # Optional dot, patch.
36 r".*$" # Suffix, including any pre- and post-release segments we don't care about.
37)
39HASH_REGEX = re.compile("^[a-fA-F0-9]{40}$")
41# SCP (Secure copy protocol) shorthand. e.g. 'git@example.com:foo/bar.git'
42SCP_REGEX = re.compile(
43 r"""^
44 # Optional user, e.g. 'git@'
45 (\w+@)?
46 # Server, e.g. 'github.com'.
47 ([^/:]+):
48 # The server-side path. e.g. 'user/project.git'. Must start with an
49 # alphanumeric character so as not to be confusable with a Windows paths
50 # like 'C:/foo/bar' or 'C:\foo\bar'.
51 (\w[^:]*)
52 $""",
53 re.VERBOSE,
54)
57def looks_like_hash(sha: str) -> bool:
58 return bool(HASH_REGEX.match(sha))
61class Git(VersionControl):
62 name = "git"
63 dirname = ".git"
64 repo_name = "clone"
65 schemes = (
66 "git+http",
67 "git+https",
68 "git+ssh",
69 "git+git",
70 "git+file",
71 )
72 # Prevent the user's environment variables from interfering with pip:
73 # https://github.com/pypa/pip/issues/1130
74 unset_environ = ("GIT_DIR", "GIT_WORK_TREE")
75 default_arg_rev = "HEAD"
77 @staticmethod
78 def get_base_rev_args(rev: str) -> list[str]:
79 return [rev]
81 @classmethod
82 def run_command(cls, *args: Any, **kwargs: Any) -> str:
83 if os.environ.get("PIP_NO_INPUT"):
84 extra_environ = kwargs.get("extra_environ", {})
85 extra_environ["GIT_TERMINAL_PROMPT"] = "0"
86 extra_environ["GIT_SSH_COMMAND"] = "ssh -oBatchMode=yes"
87 kwargs["extra_environ"] = extra_environ
88 return super().run_command(*args, **kwargs)
90 def is_immutable_rev_checkout(self, url: str, dest: str) -> bool:
91 _, rev_options = self.get_url_rev_options(hide_url(url))
92 if not rev_options.rev:
93 return False
94 if not self.is_commit_id_equal(dest, rev_options.rev):
95 # the current commit is different from rev,
96 # which means rev was something else than a commit hash
97 return False
98 # return False in the rare case rev is both a commit hash
99 # and a tag or a branch; we don't want to cache in that case
100 # because that branch/tag could point to something else in the future
101 is_tag_or_branch = bool(self.get_revision_sha(dest, rev_options.rev)[0])
102 return not is_tag_or_branch
104 def get_git_version(self) -> tuple[int, ...]:
105 version = self.run_command(
106 ["version"],
107 command_desc="git version",
108 show_stdout=False,
109 stdout_only=True,
110 )
111 match = GIT_VERSION_REGEX.match(version)
112 if not match:
113 logger.warning("Can't parse git version: %s", version)
114 return ()
115 return (int(match.group(1)), int(match.group(2)))
117 @classmethod
118 def get_current_branch(cls, location: str) -> str | None:
119 """
120 Return the current branch, or None if HEAD isn't at a branch
121 (e.g. detached HEAD).
122 """
123 # git-symbolic-ref exits with empty stdout if "HEAD" is a detached
124 # HEAD rather than a symbolic ref. In addition, the -q causes the
125 # command to exit with status code 1 instead of 128 in this case
126 # and to suppress the message to stderr.
127 args = ["symbolic-ref", "-q", "HEAD"]
128 output = cls.run_command(
129 args,
130 extra_ok_returncodes=(1,),
131 show_stdout=False,
132 stdout_only=True,
133 cwd=location,
134 )
135 ref = output.strip()
137 if ref.startswith("refs/heads/"):
138 return ref[len("refs/heads/") :]
140 return None
142 @classmethod
143 def get_revision_sha(cls, dest: str, rev: str) -> tuple[str | None, bool]:
144 """
145 Return (sha_or_none, is_branch), where sha_or_none is a commit hash
146 if the revision names a remote branch or tag, otherwise None.
148 Args:
149 dest: the repository directory.
150 rev: the revision name.
151 """
152 # Pass rev to pre-filter the list.
153 output = cls.run_command(
154 ["show-ref", rev],
155 cwd=dest,
156 show_stdout=False,
157 stdout_only=True,
158 on_returncode="ignore",
159 )
160 refs = {}
161 # NOTE: We do not use splitlines here since that would split on other
162 # unicode separators, which can be maliciously used to install a
163 # different revision.
164 for line in output.strip().split("\n"):
165 line = line.rstrip("\r")
166 if not line:
167 continue
168 try:
169 ref_sha, ref_name = line.split(" ", maxsplit=2)
170 except ValueError:
171 # Include the offending line to simplify troubleshooting if
172 # this error ever occurs.
173 raise ValueError(f"unexpected show-ref line: {line!r}")
175 refs[ref_name] = ref_sha
177 branch_ref = f"refs/remotes/origin/{rev}"
178 tag_ref = f"refs/tags/{rev}"
180 sha = refs.get(branch_ref)
181 if sha is not None:
182 return (sha, True)
184 sha = refs.get(tag_ref)
186 return (sha, False)
188 @classmethod
189 def _should_fetch(cls, dest: str, rev: str) -> bool:
190 """
191 Return true if rev is a ref or is a commit that we don't have locally.
193 Branches and tags are not considered in this method because they are
194 assumed to be always available locally (which is a normal outcome of
195 ``git clone`` and ``git fetch --tags``).
196 """
197 if rev.startswith("refs/"):
198 # Always fetch remote refs.
199 return True
201 if not looks_like_hash(rev):
202 # Git fetch would fail with abbreviated commits.
203 return False
205 if cls.has_commit(dest, rev):
206 # Don't fetch if we have the commit locally.
207 return False
209 return True
211 @classmethod
212 def resolve_revision(
213 cls, dest: str, url: HiddenText, rev_options: RevOptions
214 ) -> RevOptions:
215 """
216 Resolve a revision to a new RevOptions object with the SHA1 of the
217 branch, tag, or ref if found.
219 Args:
220 rev_options: a RevOptions object.
221 """
222 rev = rev_options.arg_rev
223 # The arg_rev property's implementation for Git ensures that the
224 # rev return value is always non-None.
225 assert rev is not None
227 sha, is_branch = cls.get_revision_sha(dest, rev)
229 if sha is not None:
230 rev_options = rev_options.make_new(sha)
231 rev_options = replace(rev_options, branch_name=(rev if is_branch else None))
233 return rev_options
235 # Do not show a warning for the common case of something that has
236 # the form of a Git commit hash.
237 if not looks_like_hash(rev):
238 logger.info(
239 "Did not find branch or tag '%s', assuming revision or ref.",
240 rev,
241 )
243 if not cls._should_fetch(dest, rev):
244 return rev_options
246 # fetch the requested revision
247 cls.run_command(
248 make_command("fetch", "-q", url, rev_options.to_args()),
249 cwd=dest,
250 )
251 # Change the revision to the SHA of the ref we fetched
252 sha = cls.get_revision(dest, rev="FETCH_HEAD")
253 rev_options = rev_options.make_new(sha)
255 return rev_options
257 @classmethod
258 def is_commit_id_equal(cls, dest: str, name: str | None) -> bool:
259 """
260 Return whether the current commit hash equals the given name.
262 Args:
263 dest: the repository directory.
264 name: a string name.
265 """
266 if not name:
267 # Then avoid an unnecessary subprocess call.
268 return False
270 return cls.get_revision(dest) == name
272 def fetch_new(
273 self, dest: str, url: HiddenText, rev_options: RevOptions, verbosity: int
274 ) -> None:
275 rev_display = rev_options.to_display()
276 logger.info("Cloning %s%s to %s", url, rev_display, display_path(dest))
277 if verbosity <= 0:
278 flags: tuple[str, ...] = ("--quiet",)
279 elif verbosity == 1:
280 flags = ()
281 else:
282 flags = ("--verbose", "--progress")
283 if self.get_git_version() >= (2, 17) and not strtobool(
284 os.environ.get("PIP_NO_PARTIAL_CLONE_FOR_BROKEN_GIT_SERVER", "no")
285 ):
286 # Git added support for partial clone in 2.17
287 # https://git-scm.com/docs/partial-clone
288 # Speeds up cloning by functioning without a complete copy of repository
289 self.run_command(
290 make_command(
291 "clone",
292 "--filter=blob:none",
293 *flags,
294 url,
295 dest,
296 )
297 )
298 else:
299 self.run_command(make_command("clone", *flags, url, dest))
301 if rev_options.rev:
302 # Then a specific revision was requested.
303 rev_options = self.resolve_revision(dest, url, rev_options)
304 branch_name = getattr(rev_options, "branch_name", None)
305 logger.debug("Rev options %s, branch_name %s", rev_options, branch_name)
306 if branch_name is None:
307 # Only do a checkout if the current commit id doesn't match
308 # the requested revision.
309 if not self.is_commit_id_equal(dest, rev_options.rev):
310 cmd_args = make_command(
311 "checkout",
312 "-q",
313 rev_options.to_args(),
314 )
315 self.run_command(cmd_args, cwd=dest)
316 elif self.get_current_branch(dest) != branch_name:
317 # Then a specific branch was requested, and that branch
318 # is not yet checked out.
319 track_branch = f"origin/{branch_name}"
320 cmd_args = [
321 "checkout",
322 "-b",
323 branch_name,
324 "--track",
325 track_branch,
326 ]
327 self.run_command(cmd_args, cwd=dest)
328 else:
329 sha = self.get_revision(dest)
330 rev_options = rev_options.make_new(sha)
332 logger.info("Resolved %s to commit %s", url, rev_options.rev)
334 #: repo may contain submodules
335 self.update_submodules(dest, verbosity=verbosity)
337 def switch(
338 self,
339 dest: str,
340 url: HiddenText,
341 rev_options: RevOptions,
342 verbosity: int = 0,
343 ) -> None:
344 self.run_command(
345 make_command("config", "remote.origin.url", url),
346 cwd=dest,
347 )
349 extra_flags = []
351 if verbosity <= 0:
352 extra_flags.append("-q")
354 cmd_args = make_command("checkout", *extra_flags, rev_options.to_args())
355 self.run_command(cmd_args, cwd=dest)
357 self.update_submodules(dest, verbosity=verbosity)
359 def update(
360 self,
361 dest: str,
362 url: HiddenText,
363 rev_options: RevOptions,
364 verbosity: int = 0,
365 ) -> None:
366 extra_flags = []
368 if verbosity <= 0:
369 extra_flags.append("-q")
371 # First fetch changes from the default remote
372 if self.get_git_version() >= (1, 9):
373 # fetch tags in addition to everything else
374 self.run_command(["fetch", "--tags", *extra_flags], cwd=dest)
375 else:
376 self.run_command(["fetch", *extra_flags], cwd=dest)
377 # Then reset to wanted revision (maybe even origin/master)
378 rev_options = self.resolve_revision(dest, url, rev_options)
379 cmd_args = make_command(
380 "reset",
381 "--hard",
382 *extra_flags,
383 rev_options.to_args(),
384 )
385 self.run_command(cmd_args, cwd=dest)
386 #: update submodules
387 self.update_submodules(dest, verbosity=verbosity)
389 @classmethod
390 def get_remote_url(cls, location: str) -> str:
391 """
392 Return URL of the first remote encountered.
394 Raises RemoteNotFoundError if the repository does not have a remote
395 url configured.
396 """
397 # We need to pass 1 for extra_ok_returncodes since the command
398 # exits with return code 1 if there are no matching lines.
399 stdout = cls.run_command(
400 ["config", "--get-regexp", r"remote\..*\.url"],
401 extra_ok_returncodes=(1,),
402 show_stdout=False,
403 stdout_only=True,
404 cwd=location,
405 )
406 remotes = stdout.splitlines()
407 try:
408 found_remote = remotes[0]
409 except IndexError:
410 raise RemoteNotFoundError
412 for remote in remotes:
413 if remote.startswith("remote.origin.url "):
414 found_remote = remote
415 break
416 url = found_remote.split(" ")[1]
417 return cls._git_remote_to_pip_url(url.strip())
419 @staticmethod
420 def _git_remote_to_pip_url(url: str) -> str:
421 """
422 Convert a remote url from what git uses to what pip accepts.
424 There are 3 legal forms **url** may take:
426 1. A fully qualified url: ssh://git@example.com/foo/bar.git
427 2. A local project.git folder: /path/to/bare/repository.git
428 3. SCP shorthand for form 1: git@example.com:foo/bar.git
430 Form 1 is output as-is. Form 2 must be converted to URI and form 3 must
431 be converted to form 1.
433 See the corresponding test test_git_remote_url_to_pip() for examples of
434 sample inputs/outputs.
435 """
436 if re.match(r"\w+://", url):
437 # This is already valid. Pass it though as-is.
438 return url
439 if os.path.exists(url):
440 # A local bare remote (git clone --mirror).
441 # Needs a file:// prefix.
442 return pathlib.Path(url).as_uri()
443 scp_match = SCP_REGEX.match(url)
444 if scp_match:
445 # Add an ssh:// prefix and replace the ':' with a '/'.
446 return scp_match.expand(r"ssh://\1\2/\3")
447 # Otherwise, bail out.
448 raise RemoteNotValidError(url)
450 @classmethod
451 def has_commit(cls, location: str, rev: str) -> bool:
452 """
453 Check if rev is a commit that is available in the local repository.
454 """
455 try:
456 cls.run_command(
457 ["rev-parse", "-q", "--verify", rev + "^{commit}"],
458 cwd=location,
459 log_failed_cmd=False,
460 )
461 except InstallationError:
462 return False
463 else:
464 return True
466 @classmethod
467 def get_revision(cls, location: str, rev: str | None = None) -> str:
468 if rev is None:
469 rev = "HEAD"
470 current_rev = cls.run_command(
471 ["rev-parse", rev],
472 show_stdout=False,
473 stdout_only=True,
474 cwd=location,
475 )
476 return current_rev.strip()
478 @classmethod
479 def get_subdirectory(cls, location: str) -> str | None:
480 """
481 Return the path to Python project root, relative to the repo root.
482 Return None if the project root is in the repo root.
483 """
484 # find the repo root
485 git_dir = cls.run_command(
486 ["rev-parse", "--git-dir"],
487 show_stdout=False,
488 stdout_only=True,
489 cwd=location,
490 ).strip()
491 if not os.path.isabs(git_dir):
492 git_dir = os.path.join(location, git_dir)
493 repo_root = os.path.abspath(os.path.join(git_dir, ".."))
494 return find_path_to_project_root_from_repo_root(location, repo_root)
496 @classmethod
497 def get_url_rev_and_auth(cls, url: str) -> tuple[str, str | None, AuthInfo]:
498 """
499 Prefixes stub URLs like 'user@hostname:user/repo.git' with 'ssh://'.
500 That's required because although they use SSH they sometimes don't
501 work with a ssh:// scheme (e.g. GitHub). But we need a scheme for
502 parsing. Hence we remove it again afterwards and return it as a stub.
503 """
504 import urllib.request
506 # Works around an apparent Git bug
507 # (see https://article.gmane.org/gmane.comp.version-control.git/146500)
508 scheme, netloc, path, query, fragment = urlsplit(url)
509 if scheme.endswith("file"):
510 initial_slashes = path[: -len(path.lstrip("/"))]
511 newpath = initial_slashes + urllib.request.url2pathname(path).replace(
512 "\\", "/"
513 ).lstrip("/")
514 after_plus = scheme.find("+") + 1
515 url = scheme[:after_plus] + urlunsplit(
516 (scheme[after_plus:], netloc, newpath, query, fragment),
517 )
519 if "://" not in url:
520 assert "file:" not in url
521 url = url.replace("git+", "git+ssh://")
522 url, rev, user_pass = super().get_url_rev_and_auth(url)
523 url = url.replace("ssh://", "")
524 else:
525 url, rev, user_pass = super().get_url_rev_and_auth(url)
527 return url, rev, user_pass
529 @classmethod
530 def update_submodules(cls, location: str, verbosity: int = 0) -> None:
531 argv = ["submodule", "update", "--init", "--recursive"]
533 if verbosity <= 0:
534 argv.append("-q")
536 if not os.path.exists(os.path.join(location, ".gitmodules")):
537 return
538 cls.run_command(
539 argv,
540 cwd=location,
541 )
543 @classmethod
544 def get_repository_root(cls, location: str) -> str | None:
545 loc = super().get_repository_root(location)
546 if loc:
547 return loc
548 try:
549 r = cls.run_command(
550 ["rev-parse", "--show-toplevel"],
551 cwd=location,
552 show_stdout=False,
553 stdout_only=True,
554 on_returncode="raise",
555 log_failed_cmd=False,
556 )
557 except BadCommand:
558 logger.debug(
559 "could not determine if %s is under git control "
560 "because git is not available",
561 location,
562 )
563 return None
564 except InstallationError:
565 return None
566 return os.path.normpath(r.rstrip("\r\n"))
568 @staticmethod
569 def should_add_vcs_url_prefix(repo_url: str) -> bool:
570 """In either https or ssh form, requirements must be prefixed with git+."""
571 return True
574vcs.register(Git)