Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/git/util.py: 50%
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
1# Copyright (C) 2008, 2009 Michael Trier (mtrier@gmail.com) and contributors
2#
3# This module is part of GitPython and is released under the
4# 3-Clause BSD License: https://opensource.org/license/bsd-3-clause/
6import sys
8__all__ = [
9 "stream_copy",
10 "join_path",
11 "to_native_path_linux",
12 "join_path_native",
13 "Stats",
14 "IndexFileSHA1Writer",
15 "IterableObj",
16 "IterableList",
17 "BlockingLockFile",
18 "LockFile",
19 "Actor",
20 "get_user_id",
21 "assure_directory_exists",
22 "RemoteProgress",
23 "CallableRemoteProgress",
24 "rmtree",
25 "unbare_repo",
26 "HIDE_WINDOWS_KNOWN_ERRORS",
27]
29if sys.platform == "win32":
30 __all__.append("to_native_path_windows")
32from abc import abstractmethod
33import contextlib
34from functools import wraps
35import getpass
36import logging
37import os
38import os.path as osp
39from pathlib import Path
40import platform
41import re
42import shutil
43import stat
44import subprocess
45import time
46from urllib.parse import urlsplit, urlunsplit
47import warnings
49# NOTE: Unused imports can be improved now that CI testing has fully resumed. Some of
50# these be used indirectly through other GitPython modules, which avoids having to write
51# gitdb all the time in their imports. They are not in __all__, at least currently,
52# because they could be removed or changed at any time, and so should not be considered
53# conceptually public to code outside GitPython. Linters of course do not like it.
54from gitdb.util import (
55 LazyMixin, # noqa: F401
56 LockedFD, # noqa: F401
57 bin_to_hex, # noqa: F401
58 file_contents_ro, # noqa: F401
59 file_contents_ro_filepath, # noqa: F401
60 hex_to_bin, # noqa: F401
61 make_sha,
62 to_bin_sha, # noqa: F401
63 to_hex_sha, # noqa: F401
64)
66# typing ---------------------------------------------------------
68from typing import (
69 Any,
70 AnyStr,
71 BinaryIO,
72 Callable,
73 Dict,
74 Generator,
75 IO,
76 Iterator,
77 List,
78 Optional,
79 Pattern,
80 Sequence,
81 Tuple,
82 TYPE_CHECKING,
83 Type,
84 TypeVar,
85 Union,
86 cast,
87 overload,
88)
90if TYPE_CHECKING:
91 from git.cmd import Git
92 from git.config import GitConfigParser, SectionConstraint
93 from git.remote import Remote
94 from git.repo.base import Repo
96from git.types import (
97 Files_TD,
98 Has_id_attribute,
99 HSH_TD,
100 Literal,
101 PathLike,
102 Protocol,
103 SupportsIndex,
104 Total_TD,
105 runtime_checkable,
106)
108# ---------------------------------------------------------------------
110T_IterableObj = TypeVar("T_IterableObj", bound=Union["IterableObj", "Has_id_attribute"], covariant=True)
111# So IterableList[Head] is subtype of IterableList[IterableObj].
112T_Actor = TypeVar("T_Actor", bound="Actor")
114_logger = logging.getLogger(__name__)
117def _read_env_flag(name: str, default: bool) -> bool:
118 """Read a boolean flag from an environment variable.
120 :return:
121 The flag, or the `default` value if absent or ambiguous.
122 """
123 try:
124 value = os.environ[name]
125 except KeyError:
126 return default
128 _logger.warning(
129 "The %s environment variable is deprecated. Its effect has never been documented and changes without warning.",
130 name,
131 )
133 adjusted_value = value.strip().lower()
135 if adjusted_value in {"", "0", "false", "no"}:
136 return False
137 if adjusted_value in {"1", "true", "yes"}:
138 return True
139 _logger.warning("%s has unrecognized value %r, treating as %r.", name, value, default)
140 return default
143def _read_win_env_flag(name: str, default: bool) -> bool:
144 """Read a boolean flag from an environment variable on Windows.
146 :return:
147 On Windows, the flag, or the `default` value if absent or ambiguous.
148 On all other operating systems, ``False``.
150 :note:
151 This only accesses the environment on Windows.
152 """
153 return sys.platform == "win32" and _read_env_flag(name, default)
156#: We need an easy way to see if Appveyor TCs start failing,
157#: so the errors marked with this var are considered "acknowledged" ones, awaiting remedy,
158#: till then, we wish to hide them.
159HIDE_WINDOWS_KNOWN_ERRORS = _read_win_env_flag("HIDE_WINDOWS_KNOWN_ERRORS", True)
160HIDE_WINDOWS_FREEZE_ERRORS = _read_win_env_flag("HIDE_WINDOWS_FREEZE_ERRORS", True)
162# { Utility Methods
164T = TypeVar("T")
167def unbare_repo(func: Callable[..., T]) -> Callable[..., T]:
168 """Methods with this decorator raise :exc:`~git.exc.InvalidGitRepositoryError` if
169 they encounter a bare repository."""
171 from .exc import InvalidGitRepositoryError
173 @wraps(func)
174 def wrapper(self: "Remote", *args: Any, **kwargs: Any) -> T:
175 if self.repo.bare:
176 raise InvalidGitRepositoryError("Method '%s' cannot operate on bare repositories" % func.__name__)
177 # END bare method
178 return func(self, *args, **kwargs)
180 # END wrapper
182 return wrapper
185@contextlib.contextmanager
186def cwd(new_dir: PathLike) -> Generator[PathLike, None, None]:
187 """Context manager to temporarily change directory.
189 This is similar to :func:`contextlib.chdir` introduced in Python 3.11, but the
190 context manager object returned by a single call to this function is not reentrant.
191 """
192 old_dir = os.getcwd()
193 os.chdir(new_dir)
194 try:
195 yield new_dir
196 finally:
197 os.chdir(old_dir)
200@contextlib.contextmanager
201def patch_env(name: str, value: str) -> Generator[None, None, None]:
202 """Context manager to temporarily patch an environment variable."""
203 old_value = os.getenv(name)
204 os.environ[name] = value
205 try:
206 yield
207 finally:
208 if old_value is None:
209 del os.environ[name]
210 else:
211 os.environ[name] = old_value
214def rmtree(path: PathLike) -> None:
215 """Remove the given directory tree recursively.
217 :note:
218 We use :func:`shutil.rmtree` but adjust its behaviour to see whether files that
219 couldn't be deleted are read-only. Windows will not remove them in that case.
220 """
222 def handler(function: Callable[[str], Any], path: str, _excinfo: Any) -> None:
223 """Callback for :func:`shutil.rmtree`.
225 This works as either a ``onexc`` or ``onerror`` style callback.
226 """
227 # Is the error an access error?
228 os.chmod(path, stat.S_IWUSR)
230 try:
231 function(path)
232 except PermissionError as ex:
233 if HIDE_WINDOWS_KNOWN_ERRORS:
234 from unittest import SkipTest
236 raise SkipTest(f"FIXME: fails with: PermissionError\n {ex}") from ex
237 raise
239 if sys.platform != "win32":
240 shutil.rmtree(path)
241 elif sys.version_info >= (3, 12):
242 shutil.rmtree(path, onexc=handler)
243 else:
244 shutil.rmtree(path, onerror=handler)
247def rmfile(path: PathLike) -> None:
248 """Ensure file deleted also on *Windows* where read-only files need special
249 treatment."""
250 if osp.isfile(path):
251 if sys.platform == "win32":
252 os.chmod(path, 0o777)
253 os.remove(path)
256def stream_copy(source: BinaryIO, destination: BinaryIO, chunk_size: int = 512 * 1024) -> int:
257 """Copy all data from the `source` stream into the `destination` stream in chunks
258 of size `chunk_size`.
260 :return:
261 Number of bytes written
262 """
263 br = 0
264 while True:
265 chunk = source.read(chunk_size)
266 destination.write(chunk)
267 br += len(chunk)
268 if len(chunk) < chunk_size:
269 break
270 # END reading output stream
271 return br
274def join_path(a: PathLike, *p: PathLike) -> PathLike:
275 R"""Join path tokens together similar to osp.join, but always use ``/`` instead of
276 possibly ``\`` on Windows."""
277 path = os.fspath(a)
278 for b in p:
279 b = os.fspath(b)
280 if not b:
281 continue
282 if b.startswith("/"):
283 path += b[1:]
284 elif path == "" or path.endswith("/"):
285 path += b
286 else:
287 path += "/" + b
288 # END for each path token to add
289 return path
292if sys.platform == "win32":
294 def to_native_path_windows(path: PathLike) -> str:
295 path = os.fspath(path)
296 return path.replace("/", "\\")
298 def to_native_path_linux(path: PathLike) -> str:
299 path = os.fspath(path)
300 return path.replace("\\", "/")
302 to_native_path = to_native_path_windows
303else:
304 # No need for any work on Linux.
305 def to_native_path_linux(path: PathLike) -> str:
306 return os.fspath(path)
308 to_native_path = to_native_path_linux
311def join_path_native(a: PathLike, *p: PathLike) -> PathLike:
312 R"""Like :func:`join_path`, but makes sure an OS native path is returned.
314 This is only needed to play it safe on Windows and to ensure nice paths that only
315 use ``\``.
316 """
317 return to_native_path(join_path(a, *p))
320def _is_path_rooted(path: PathLike) -> bool:
321 r"""Whether ``path`` has a root, including one encoded in a UNC drive.
323 On Windows, ``\directory`` is rooted on the current drive without being
324 absolute, while ``C:\directory`` has both a drive and a root. In contrast,
325 ``directory`` and the drive-relative ``C:directory`` have no root.
326 UNC paths are rooted: ``\\server\share`` stores the share in the drive
327 returned by :func:`os.path.splitdrive`, while ``\\server\share\directory``
328 additionally has a rooted tail.
329 On POSIX, which has no drive concept, this simply distinguishes absolute
330 paths such as ``/directory`` from relative paths such as ``directory``.
331 """
332 drive, tail = osp.splitdrive(os.fspath(path))
333 separators = (os.sep,) if os.altsep is None else (os.sep, os.altsep)
334 return tail.startswith(separators) or drive.startswith(separators)
337def _to_relative_path(root: PathLike, path: PathLike) -> str:
338 r"""Return a normalized Git-style path confined to ``root``.
340 A Windows path such as ``\directory`` is rooted but not absolute. Resolve it
341 against the drive of ``root`` rather than treating it as relative to ``root``.
342 Drive-relative paths such as ``C:directory`` are rejected because their meaning
343 depends on process-global per-drive state.
345 For example, with ``root`` set to ``C:\repo`` on Windows:
347 * ``directory\file`` -> ``directory/file``
348 * ``directory\`` -> ``directory/``
349 * ``C:\repo\directory\file`` -> ``directory/file``
350 * ``\repo\directory\file`` -> ``directory/file``
351 * ``C:directory\file`` -> :exc:`ValueError`
352 * ``C:\other\file`` -> :exc:`ValueError`
354 On POSIX, ``/repo/directory/file`` under ``/repo`` similarly becomes
355 ``directory/file``. A trailing separator is preserved as a Git-style ``/``.
356 """
357 path_str = os.fspath(path)
358 if not path_str:
359 return path_str
361 drive, _tail = osp.splitdrive(path_str)
362 rooted = _is_path_rooted(path_str)
363 if drive and not rooted:
364 raise ValueError("Drive-relative path %r is not supported" % path_str)
366 root_abs = osp.abspath(os.fspath(root))
367 path_abs = osp.abspath(osp.join(root_abs, path_str))
368 try:
369 common_path = osp.commonpath([root_abs, path_abs])
370 except ValueError as e:
371 raise ValueError("Path %r is not in repository at %r" % (path_str, root_abs)) from e
372 if common_path != root_abs:
373 raise ValueError("Path %r is not in repository at %r" % (path_str, root_abs))
375 relative_path = to_native_path_linux(osp.relpath(path_abs, root_abs))
376 separators = (os.sep,) if os.altsep is None else (os.sep, os.altsep)
377 if path_str.endswith(separators) and relative_path != "." and not relative_path.endswith("/"):
378 relative_path += "/"
379 return relative_path
382def assure_directory_exists(path: PathLike, is_file: bool = False) -> bool:
383 """Make sure that the directory pointed to by path exists.
385 :param is_file:
386 If ``True``, `path` is assumed to be a file and handled correctly.
387 Otherwise it must be a directory.
389 :return:
390 ``True`` if the directory was created, ``False`` if it already existed.
391 """
392 if is_file:
393 path = osp.dirname(path)
394 # END handle file
395 if not osp.isdir(path):
396 os.makedirs(path, exist_ok=True)
397 return True
398 return False
401def _get_exe_extensions() -> Sequence[str]:
402 PATHEXT = os.environ.get("PATHEXT", None)
403 if PATHEXT:
404 return tuple(p.upper() for p in PATHEXT.split(os.pathsep))
405 elif sys.platform == "win32":
406 return (".BAT", ".COM", ".EXE")
407 else:
408 return ()
411def py_where(program: str, path: Optional[PathLike] = None) -> List[str]:
412 """Perform a path search to assist :func:`is_cygwin_git`.
414 This is not robust for general use. It is an implementation detail of
415 :func:`is_cygwin_git`. When a search following all shell rules is needed,
416 :func:`shutil.which` can be used instead.
418 :note:
419 Neither this function nor :func:`shutil.which` will predict the effect of an
420 executable search on a native Windows system due to a :class:`subprocess.Popen`
421 call without ``shell=True``, because shell and non-shell executable search on
422 Windows differ considerably.
423 """
424 # From: http://stackoverflow.com/a/377028/548792
425 winprog_exts = _get_exe_extensions()
427 def is_exec(fpath: str) -> bool:
428 return (
429 osp.isfile(fpath)
430 and os.access(fpath, os.X_OK)
431 and (
432 sys.platform != "win32" or not winprog_exts or any(fpath.upper().endswith(ext) for ext in winprog_exts)
433 )
434 )
436 progs = []
437 if not path:
438 path = os.environ["PATH"]
439 for folder in os.fspath(path).split(os.pathsep):
440 folder = folder.strip('"')
441 if folder:
442 exe_path = osp.join(folder, program)
443 for f in [exe_path] + ["%s%s" % (exe_path, e) for e in winprog_exts]:
444 if is_exec(f):
445 progs.append(f)
446 return progs
449def _cygexpath(drive: Optional[str], path: str, expand_vars: bool = True) -> str:
450 if osp.isabs(path) and not drive:
451 # Invoked from `cygpath()` directly with `D:Apps\123`?
452 # It's an error, leave it alone just slashes)
453 p = path # convert to str if AnyPath given
454 else:
455 p = path and osp.normpath(osp.expandvars(osp.expanduser(path)) if expand_vars else path)
456 if osp.isabs(p):
457 if drive:
458 # Confusing, maybe a remote system should expand vars.
459 p = path
460 else:
461 p = cygpath(p)
462 elif drive:
463 p = "/proc/cygdrive/%s/%s" % (drive.lower(), p)
464 p_str = os.fspath(p) # ensure it is a str and not AnyPath
465 return p_str.replace("\\", "/")
468_cygpath_parsers: Tuple[Tuple[Pattern[str], Callable[..., str], bool], ...] = (
469 # See: https://msdn.microsoft.com/en-us/library/windows/desktop/aa365247(v=vs.85).aspx
470 # and: https://www.cygwin.com/cygwin-ug-net/using.html#unc-paths
471 (
472 re.compile(r"\\\\\?\\UNC\\([^\\]+)\\([^\\]+)(?:\\(.*))?"),
473 (lambda server, share, rest_path: "//%s/%s/%s" % (server, share, rest_path.replace("\\", "/"))),
474 False,
475 ),
476 (re.compile(r"\\\\\?\\(\w):[/\\](.*)"), (_cygexpath), False),
477 (re.compile(r"(\w):[/\\](.*)"), (_cygexpath), False),
478 (re.compile(r"file:(.*)", re.I), (lambda rest_path: rest_path), True),
479 (re.compile(r"(\w{2,}:.*)"), (lambda url: url), False), # remote URL, do nothing
480)
483def cygpath(path: str, expand_vars: bool = True) -> str:
484 """Use :meth:`git.cmd.Git.polish_url` instead, that works on any environment."""
485 path = os.fspath(path) # Ensure is str and not AnyPath.
486 # Fix to use Paths when 3.5 dropped. Or to be just str if only for URLs?
487 if not path.startswith(("/cygdrive", "//", "/proc/cygdrive")):
488 for regex, parser, recurse in _cygpath_parsers:
489 match = regex.match(path)
490 if match:
491 if parser is _cygexpath:
492 path = parser(*match.groups(), expand_vars=expand_vars)
493 else:
494 path = parser(*match.groups())
495 if recurse:
496 path = cygpath(path, expand_vars=expand_vars)
497 break
498 else:
499 path = _cygexpath(None, path, expand_vars=expand_vars)
501 return path
504_decygpath_regex = re.compile(r"(?:/proc)?/cygdrive/(\w)(/.*)?")
507def decygpath(path: PathLike) -> str:
508 path = os.fspath(path)
509 m = _decygpath_regex.match(path)
510 if m:
511 drive, rest_path = m.groups()
512 path = "%s:%s" % (drive.upper(), rest_path or "")
514 return path.replace("/", "\\")
517#: Store boolean flags denoting if a specific Git executable
518#: is from a Cygwin installation (since `cache_lru()` unsupported on PY2).
519_is_cygwin_cache: Dict[str, Optional[bool]] = {}
522def _is_cygwin_git(git_executable: str) -> bool:
523 is_cygwin = _is_cygwin_cache.get(git_executable) # type: Optional[bool]
524 if is_cygwin is None:
525 is_cygwin = False
526 try:
527 git_dir = osp.dirname(git_executable)
528 if not git_dir:
529 res = py_where(git_executable)
530 git_dir = osp.dirname(res[0]) if res else ""
532 # Just a name given, not a real path.
533 uname_cmd = osp.join(git_dir, "uname")
535 if not (Path(uname_cmd).is_file() and os.access(uname_cmd, os.X_OK)):
536 _logger.debug(f"Failed checking if running in CYGWIN: {uname_cmd} is not an executable")
537 _is_cygwin_cache[git_executable] = is_cygwin
538 return is_cygwin
540 process = subprocess.Popen([uname_cmd], stdout=subprocess.PIPE, universal_newlines=True)
541 uname_out, _ = process.communicate()
542 # retcode = process.poll()
543 is_cygwin = "CYGWIN" in uname_out
544 except Exception as ex:
545 _logger.debug("Failed checking if running in CYGWIN due to: %r", ex)
546 _is_cygwin_cache[git_executable] = is_cygwin
548 return is_cygwin
551@overload
552def is_cygwin_git(git_executable: None) -> Literal[False]: ...
555@overload
556def is_cygwin_git(git_executable: PathLike) -> bool: ...
559def is_cygwin_git(git_executable: Union[None, PathLike]) -> bool:
560 # TODO: when py3.7 support is dropped, use the new interpolation f"{variable=}"
561 _logger.debug(f"sys.platform={sys.platform!r}, git_executable={git_executable!r}")
562 if sys.platform != "cygwin":
563 return False
564 elif git_executable is None:
565 return False
566 else:
567 return _is_cygwin_git(str(git_executable))
570def get_user_id() -> str:
571 """:return: String identifying the currently active system user as ``name@node``"""
572 return "%s@%s" % (getpass.getuser(), platform.node())
575def finalize_process(proc: Union["subprocess.Popen[Any]", "Git.AutoInterrupt"], **kwargs: Any) -> None:
576 """Wait for the process (clone, fetch, pull or push) and handle its errors
577 accordingly."""
578 # TODO: No close proc-streams??
579 proc.wait(**kwargs)
582@overload
583def expand_path(p: None, expand_vars: bool = ...) -> None: ...
586@overload
587def expand_path(p: PathLike, expand_vars: bool = ...) -> Optional[PathLike]:
588 # TODO: Support for Python 3.5 has been dropped, so these overloads can be improved.
589 ...
592def expand_path(p: Union[None, PathLike], expand_vars: bool = True) -> Optional[PathLike]:
593 if p is None:
594 return None
595 try:
596 if isinstance(p, Path):
597 return p.resolve()
598 expanded_path = osp.expanduser(os.fspath(p))
599 if expand_vars:
600 expanded_path = osp.expandvars(expanded_path)
601 return osp.normpath(osp.abspath(expanded_path))
602 except Exception:
603 return None
606def remove_password_if_present(cmdline: Sequence[str]) -> List[str]:
607 """Redact credentials in URLs and HTTP Authorization extra headers in a command line.
609 If nothing is found, this just returns the command line as-is.
611 This should be used for every log line that print a command line, as well as
612 exception messages.
613 """
614 new_cmdline = []
615 for index, to_parse in enumerate(cmdline):
616 new_cmdline.append(to_parse)
617 config_key, separator, header = to_parse.partition("=")
618 header_name, colon, _ = header.partition(":")
619 if (
620 separator
621 and colon
622 and config_key.lower().endswith(".extraheader")
623 and header_name.strip().lower() == "authorization"
624 ):
625 new_cmdline[index] = "%s%s%s%s *****" % (config_key, separator, header_name, colon)
626 continue
627 try:
628 url = urlsplit(to_parse)
629 # Remove password from the URL if present.
630 if url.password is None and url.username is None:
631 continue
633 if url.password is not None:
634 url = url._replace(netloc=url.netloc.replace(url.password, "*****"))
635 if url.username is not None:
636 url = url._replace(netloc=url.netloc.replace(url.username, "*****"))
637 new_cmdline[index] = urlunsplit(url)
638 except ValueError:
639 # This is not a valid URL.
640 continue
641 return new_cmdline
644# } END utilities
646# { Classes
649class RemoteProgress:
650 """Handler providing an interface to parse progress information emitted by
651 :manpage:`git-push(1)` and :manpage:`git-fetch(1)` and to dispatch callbacks
652 allowing subclasses to react to the progress."""
654 _num_op_codes: int = 9
655 (
656 BEGIN,
657 END,
658 COUNTING,
659 COMPRESSING,
660 WRITING,
661 RECEIVING,
662 RESOLVING,
663 FINDING_SOURCES,
664 CHECKING_OUT,
665 ) = [1 << x for x in range(_num_op_codes)]
666 STAGE_MASK = BEGIN | END
667 OP_MASK = ~STAGE_MASK
669 DONE_TOKEN = "done."
670 TOKEN_SEPARATOR = ", "
672 __slots__ = (
673 "_cur_line",
674 "_seen_ops",
675 "error_lines", # Lines that started with 'error:' or 'fatal:'.
676 "other_lines", # Lines not denoting progress (i.e.g. push-infos).
677 )
678 re_op_absolute = re.compile(r"(remote: )?([\w\s]+):\s+()(\d+)()(.*)")
679 re_op_relative = re.compile(r"(remote: )?([\w\s]+):\s+(\d+)% \((\d+)/(\d+)\)(.*)")
681 def __init__(self) -> None:
682 self._seen_ops: List[int] = []
683 self._cur_line: Optional[str] = None
684 self.error_lines: List[str] = []
685 self.other_lines: List[str] = []
687 def _parse_progress_line(self, line: AnyStr) -> Optional[object]:
688 """Parse progress information from the given line as retrieved by
689 :manpage:`git-push(1)` or :manpage:`git-fetch(1)`.
691 - Lines that do not contain progress info are stored in :attr:`other_lines`.
692 - Lines that seem to contain an error (i.e. start with ``error:`` or ``fatal:``)
693 are stored in :attr:`error_lines`.
695 The base implementation returns ``None``. Subclasses may return another
696 value for compatibility with existing overrides, but callers should treat
697 the return value as unspecified.
698 """
699 # handle
700 # Counting objects: 4, done.
701 # Compressing objects: 50% (1/2)
702 # Compressing objects: 100% (2/2)
703 # Compressing objects: 100% (2/2), done.
704 if isinstance(line, bytes): # mypy argues about ternary assignment.
705 line_str = line.decode("utf-8")
706 else:
707 line_str = line
708 self._cur_line = line_str
710 if self._cur_line.startswith(("error:", "fatal:")):
711 self.error_lines.append(self._cur_line)
712 return None
714 cur_count, max_count = None, None
715 match = self.re_op_relative.match(line_str)
716 if match is None:
717 match = self.re_op_absolute.match(line_str)
719 if not match:
720 self.line_dropped(line_str)
721 self.other_lines.append(line_str)
722 return None
723 # END could not get match
725 op_code = 0
726 _remote, op_name, _percent, cur_count, max_count, message = match.groups()
728 # Get operation ID.
729 if op_name == "Counting objects":
730 op_code |= self.COUNTING
731 elif op_name == "Compressing objects":
732 op_code |= self.COMPRESSING
733 elif op_name == "Writing objects":
734 op_code |= self.WRITING
735 elif op_name == "Receiving objects":
736 op_code |= self.RECEIVING
737 elif op_name == "Resolving deltas":
738 op_code |= self.RESOLVING
739 elif op_name == "Finding sources":
740 op_code |= self.FINDING_SOURCES
741 elif op_name == "Checking out files":
742 op_code |= self.CHECKING_OUT
743 else:
744 # Note: On Windows it can happen that partial lines are sent.
745 # Hence we get something like "CompreReceiving objects", which is
746 # a blend of "Compressing objects" and "Receiving objects".
747 # This can't really be prevented, so we drop the line verbosely
748 # to make sure we get informed in case the process spits out new
749 # commands at some point.
750 self.line_dropped(line_str)
751 # Note: Don't add this line to the other lines, as we have to silently
752 # drop it.
753 return None
754 # END handle op code
756 # Figure out stage.
757 if op_code not in self._seen_ops:
758 self._seen_ops.append(op_code)
759 op_code |= self.BEGIN
760 # END begin opcode
762 if message is None:
763 message = ""
764 # END message handling
766 message = message.strip()
767 if message.endswith(self.DONE_TOKEN):
768 op_code |= self.END
769 message = message[: -len(self.DONE_TOKEN)]
770 # END end message handling
771 message = message.strip(self.TOKEN_SEPARATOR)
773 self.update(
774 op_code,
775 cur_count and float(cur_count),
776 max_count and float(max_count),
777 message,
778 )
779 return None
781 def new_message_handler(self) -> Callable[[str], None]:
782 """
783 :return:
784 A progress handler suitable for :func:`~git.cmd.handle_process_output`,
785 passing lines on to this progress handler in a suitable format.
786 """
788 def handler(line: AnyStr) -> None:
789 self._parse_progress_line(line.rstrip())
791 # END handler
793 return handler
795 def line_dropped(self, line: str) -> None:
796 """Called whenever a line could not be understood and was therefore dropped."""
797 pass
799 def update(
800 self,
801 op_code: int,
802 cur_count: Union[str, float],
803 max_count: Union[str, float, None] = None,
804 message: str = "",
805 ) -> None:
806 """Called whenever the progress changes.
808 :param op_code:
809 Integer allowing to be compared against Operation IDs and stage IDs.
811 Stage IDs are :const:`BEGIN` and :const:`END`. :const:`BEGIN` will only be
812 set once for each Operation ID as well as :const:`END`. It may be that
813 :const:`BEGIN` and :const:`END` are set at once in case only one progress
814 message was emitted due to the speed of the operation. Between
815 :const:`BEGIN` and :const:`END`, none of these flags will be set.
817 Operation IDs are all held within the :const:`OP_MASK`. Only one Operation
818 ID will be active per call.
820 :param cur_count:
821 Current absolute count of items.
823 :param max_count:
824 The maximum count of items we expect. It may be ``None`` in case there is no
825 maximum number of items or if it is (yet) unknown.
827 :param message:
828 In case of the :const:`WRITING` operation, it contains the amount of bytes
829 transferred. It may possibly be used for other purposes as well.
831 :note:
832 You may read the contents of the current line in
833 :attr:`self._cur_line <_cur_line>`.
834 """
835 pass
838class CallableRemoteProgress(RemoteProgress):
839 """A :class:`RemoteProgress` implementation forwarding updates to any callable.
841 :note:
842 Like direct instances of :class:`RemoteProgress`, instances of this
843 :class:`CallableRemoteProgress` class are not themselves directly callable.
844 Rather, instances of this class wrap a callable and forward to it. This should
845 therefore not be confused with :class:`git.types.CallableProgress`.
846 """
848 __slots__ = ("_callable",)
850 def __init__(self, fn: Callable[..., Any]) -> None:
851 self._callable = fn
852 super().__init__()
854 def update(self, *args: Any, **kwargs: Any) -> None:
855 self._callable(*args, **kwargs)
858class _DeprecatedActorNameEmailRegex:
859 _pattern = re.compile(r"(.*) <(.*?)>")
861 def __get__(self, _instance: Any, _owner: Any) -> Pattern[str]:
862 warnings.warn(
863 "Actor.name_email_regex is deprecated and will be removed in GitPython 4.0.0 because searching long "
864 "malformed strings with it can take quadratic time. Use Actor.from_string() to parse actor identities, "
865 "or Actor(name, email) when the fields are already separate.",
866 DeprecationWarning,
867 stacklevel=2,
868 )
869 return self._pattern
872class Actor:
873 """Actors hold information about a person acting on the repository. They can be
874 committers and authors or anything with a name and an email as mentioned in the git
875 log entries."""
877 name_email_regex = _DeprecatedActorNameEmailRegex()
879 # ENVIRONMENT VARIABLES
880 # These are read when creating new commits.
881 env_author_name = "GIT_AUTHOR_NAME"
882 env_author_email = "GIT_AUTHOR_EMAIL"
883 env_committer_name = "GIT_COMMITTER_NAME"
884 env_committer_email = "GIT_COMMITTER_EMAIL"
886 # CONFIGURATION KEYS
887 conf_name = "name"
888 conf_email = "email"
890 __slots__ = ("name", "email")
892 def __init__(self, name: Optional[str], email: Optional[str]) -> None:
893 self.name = name
894 self.email = email
896 def __eq__(self, other: Any) -> bool:
897 return self.name == other.name and self.email == other.email
899 def __ne__(self, other: Any) -> bool:
900 return not (self == other)
902 def __hash__(self) -> int:
903 return hash((self.name, self.email))
905 def __str__(self) -> str:
906 return self.name if self.name else ""
908 def __repr__(self) -> str:
909 return '<git.Actor "%s <%s>">' % (self.name, self.email)
911 @classmethod
912 def from_string(cls: Type[T_Actor], string: str) -> T_Actor:
913 """Create an :class:`Actor` from a string.
915 :param string:
916 The string, which is expected to be in regular git format::
918 John Doe <jdoe@example.com>
920 :return:
921 :class:`Actor`
922 """
923 line = string.partition("\n")[0]
924 left_bracket = line.find("<")
925 right_bracket = line.find(">", left_bracket + 1)
926 if left_bracket >= 0 and right_bracket >= 0:
927 return cls(line[:left_bracket].rstrip(), line[left_bracket + 1 : right_bracket])
929 # Assume the best and use the whole string as name.
930 return cls(string, None)
932 _from_string = from_string
934 @classmethod
935 def _main_actor(
936 cls,
937 env_name: str,
938 env_email: str,
939 config_reader: Union[None, "GitConfigParser", "SectionConstraint[GitConfigParser]"] = None,
940 ) -> "Actor":
941 actor = Actor("", "")
942 user_id = None # We use this to avoid multiple calls to getpass.getuser().
944 def default_email() -> str:
945 nonlocal user_id
946 if not user_id:
947 user_id = get_user_id()
948 return user_id
950 def default_name() -> str:
951 return default_email().split("@")[0]
953 for attr, evar, cvar, default in (
954 ("name", env_name, cls.conf_name, default_name),
955 ("email", env_email, cls.conf_email, default_email),
956 ):
957 try:
958 val = os.environ[evar]
959 setattr(actor, attr, val)
960 except KeyError:
961 if config_reader is not None:
962 try:
963 val = config_reader.get("user", cvar)
964 except Exception:
965 val = default()
966 setattr(actor, attr, val)
967 # END config-reader handling
968 if not getattr(actor, attr):
969 setattr(actor, attr, default())
970 # END handle name
971 # END for each item to retrieve
972 return actor
974 @classmethod
975 def committer(
976 cls, config_reader: Union[None, "GitConfigParser", "SectionConstraint[GitConfigParser]"] = None
977 ) -> "Actor":
978 """
979 :return:
980 :class:`Actor` instance corresponding to the configured committer. It
981 behaves similar to the git implementation, such that the environment will
982 override configuration values of `config_reader`. If no value is set at all,
983 it will be generated.
985 :param config_reader:
986 ConfigReader to use to retrieve the values from in case they are not set in
987 the environment.
988 """
989 return cls._main_actor(cls.env_committer_name, cls.env_committer_email, config_reader)
991 @classmethod
992 def author(
993 cls, config_reader: Union[None, "GitConfigParser", "SectionConstraint[GitConfigParser]"] = None
994 ) -> "Actor":
995 """Same as :meth:`committer`, but defines the main author. It may be specified
996 in the environment, but defaults to the committer."""
997 return cls._main_actor(cls.env_author_name, cls.env_author_email, config_reader)
1000class Stats:
1001 """Represents stat information as presented by git at the end of a merge. It is
1002 created from the output of a diff operation.
1004 Example::
1006 c = Commit( sha1 )
1007 s = c.stats
1008 s.total # full-stat-dict
1009 s.files # dict( filepath : stat-dict )
1011 ``stat-dict``
1013 A dictionary with the following keys and values::
1015 deletions = number of deleted lines as int
1016 insertions = number of inserted lines as int
1017 lines = total number of lines changed as int, or deletions + insertions
1018 change_type = type of change as str, A|C|D|M|R|T|U|X|B
1020 ``full-stat-dict``
1022 In addition to the items in the stat-dict, it features additional information::
1024 files = number of changed files as int
1025 """
1027 __slots__ = ("total", "files")
1029 def __init__(self, total: Total_TD, files: Dict[PathLike, Files_TD]) -> None:
1030 self.total = total
1031 self.files = files
1033 @classmethod
1034 def _list_from_string(cls, repo: "Repo", text: str) -> "Stats":
1035 """Create a :class:`Stats` object from output retrieved by
1036 :manpage:`git-diff(1)`.
1038 :return:
1039 :class:`git.Stats`
1040 """
1042 hsh: HSH_TD = {
1043 "total": {"insertions": 0, "deletions": 0, "lines": 0, "files": 0},
1044 "files": {},
1045 }
1046 for line in text.splitlines():
1047 (change_type, raw_insertions, raw_deletions, filename) = line.split("\t")
1048 insertions = raw_insertions != "-" and int(raw_insertions) or 0
1049 deletions = raw_deletions != "-" and int(raw_deletions) or 0
1050 hsh["total"]["insertions"] += insertions
1051 hsh["total"]["deletions"] += deletions
1052 hsh["total"]["lines"] += insertions + deletions
1053 hsh["total"]["files"] += 1
1054 files_dict: Files_TD = {
1055 "insertions": insertions,
1056 "deletions": deletions,
1057 "lines": insertions + deletions,
1058 "change_type": change_type,
1059 }
1060 hsh["files"][filename.strip()] = files_dict
1061 return Stats(hsh["total"], hsh["files"])
1064class IndexFileSHA1Writer:
1065 """Wrapper around a file-like object that remembers the SHA1 of the data written to
1066 it. It will write a sha when the stream is closed or if asked for explicitly using
1067 :meth:`write_sha`.
1069 Only useful to the index file.
1071 :note:
1072 Based on the dulwich project.
1073 """
1075 __slots__ = ("f", "sha1")
1077 def __init__(self, f: IO[bytes]) -> None:
1078 self.f = f
1079 self.sha1 = make_sha(b"")
1081 def write(self, data: bytes) -> int:
1082 self.sha1.update(data)
1083 return self.f.write(data)
1085 def write_sha(self) -> bytes:
1086 sha = self.sha1.digest()
1087 self.f.write(sha)
1088 return sha
1090 def close(self) -> bytes:
1091 sha = self.write_sha()
1092 self.f.close()
1093 return sha
1095 def tell(self) -> int:
1096 return self.f.tell()
1099class LockFile:
1100 """Provides methods to obtain, check for, and release a file based lock which
1101 should be used to handle concurrent access to the same file.
1103 As we are a utility class to be derived from, we only use protected methods.
1105 Locks will automatically be released on destruction.
1106 """
1108 __slots__ = ("_file_path", "_owns_lock")
1110 def __init__(self, file_path: PathLike) -> None:
1111 self._file_path = file_path
1112 self._owns_lock = False
1114 def __del__(self) -> None:
1115 self._release_lock()
1117 def _lock_file_path(self) -> str:
1118 """:return: Path to lockfile"""
1119 return "%s.lock" % (self._file_path)
1121 def _has_lock(self) -> bool:
1122 """
1123 :return:
1124 True if we have a lock and if the lockfile still exists
1126 :raise AssertionError:
1127 If our lock-file does not exist.
1128 """
1129 return self._owns_lock
1131 def _obtain_lock_or_raise(self) -> None:
1132 """Create a lock file as flag for other instances, mark our instance as
1133 lock-holder.
1135 :raise IOError:
1136 If a lock was already present or a lock file could not be written.
1137 """
1138 if self._has_lock():
1139 return
1140 lock_file = self._lock_file_path()
1141 if osp.isfile(lock_file):
1142 raise IOError(
1143 "Lock for file %r did already exist, delete %r in case the lock is illegal"
1144 % (self._file_path, lock_file)
1145 )
1147 try:
1148 with open(lock_file, mode="w"):
1149 pass
1150 except OSError as e:
1151 raise IOError(str(e)) from e
1153 self._owns_lock = True
1155 def _obtain_lock(self) -> None:
1156 """The default implementation will raise if a lock cannot be obtained.
1158 Subclasses may override this method to provide a different implementation.
1159 """
1160 return self._obtain_lock_or_raise()
1162 def _release_lock(self) -> None:
1163 """Release our lock if we have one."""
1164 if not self._has_lock():
1165 return
1167 # If someone removed our file beforehand, lets just flag this issue instead of
1168 # failing, to make it more usable.
1169 lfp = self._lock_file_path()
1170 try:
1171 rmfile(lfp)
1172 except OSError:
1173 pass
1174 self._owns_lock = False
1177class BlockingLockFile(LockFile):
1178 """The lock file will block until a lock could be obtained, or fail after a
1179 specified timeout.
1181 :note:
1182 If the directory containing the lock was removed, an exception will be raised
1183 during the blocking period, preventing hangs as the lock can never be obtained.
1184 """
1186 __slots__ = ("_check_interval", "_max_block_time")
1188 def __init__(
1189 self,
1190 file_path: PathLike,
1191 check_interval_s: float = 0.3,
1192 max_block_time_s: int = sys.maxsize,
1193 ) -> None:
1194 """Configure the instance.
1196 :param check_interval_s:
1197 Period of time to sleep until the lock is checked the next time.
1198 By default, it waits a nearly unlimited time.
1200 :param max_block_time_s:
1201 Maximum amount of seconds we may lock.
1202 """
1203 super().__init__(file_path)
1204 self._check_interval = check_interval_s
1205 self._max_block_time = max_block_time_s
1207 def _obtain_lock(self) -> None:
1208 """This method blocks until it obtained the lock, or raises :exc:`IOError` if it
1209 ran out of time or if the parent directory was not available anymore.
1211 If this method returns, you are guaranteed to own the lock.
1212 """
1213 starttime = time.time()
1214 maxtime = starttime + float(self._max_block_time)
1215 while True:
1216 try:
1217 super()._obtain_lock()
1218 except IOError as e:
1219 # synity check: if the directory leading to the lockfile is not
1220 # readable anymore, raise an exception
1221 curtime = time.time()
1222 if not osp.isdir(osp.dirname(self._lock_file_path())):
1223 msg = "Directory containing the lockfile %r was not readable anymore after waiting %g seconds" % (
1224 self._lock_file_path(),
1225 curtime - starttime,
1226 )
1227 raise IOError(msg) from e
1228 # END handle missing directory
1230 if curtime >= maxtime:
1231 msg = "Waited %g seconds for lock at %r" % (
1232 maxtime - starttime,
1233 self._lock_file_path(),
1234 )
1235 raise IOError(msg) from e
1236 # END abort if we wait too long
1237 time.sleep(self._check_interval)
1238 else:
1239 break
1240 # END endless loop
1243class IterableList(List[T_IterableObj]): # type: ignore[type-var]
1244 """List of iterable objects allowing to query an object by id or by named index::
1246 heads = repo.heads
1247 heads.master
1248 heads['master']
1249 heads[0]
1251 Iterable parent objects:
1253 * :class:`Commit <git.objects.Commit>`
1254 * :class:`Submodule <git.objects.submodule.base.Submodule>`
1255 * :class:`Reference <git.refs.reference.Reference>`
1256 * :class:`FetchInfo <git.remote.FetchInfo>`
1257 * :class:`PushInfo <git.remote.PushInfo>`
1259 Iterable via inheritance:
1261 * :class:`Head <git.refs.head.Head>`
1262 * :class:`TagReference <git.refs.tag.TagReference>`
1263 * :class:`RemoteReference <git.refs.remote.RemoteReference>`
1265 This requires an ``id_attribute`` name to be set which will be queried from its
1266 contained items to have a means for comparison.
1268 A prefix can be specified which is to be used in case the id returned by the items
1269 always contains a prefix that does not matter to the user, so it can be left out.
1270 """
1272 __slots__ = ("_id_attr", "_prefix")
1274 def __new__(cls, id_attr: str, prefix: str = "") -> "IterableList[T_IterableObj]":
1275 return super().__new__(cls)
1277 def __init__(self, id_attr: str, prefix: str = "") -> None:
1278 super().__init__()
1279 self._id_attr = id_attr
1280 self._prefix = prefix
1282 def __contains__(self, attr: object) -> bool:
1283 # First try identity match for performance.
1284 try:
1285 rval = list.__contains__(self, attr)
1286 if rval:
1287 return rval
1288 except (AttributeError, TypeError):
1289 pass
1290 # END handle match
1292 # Otherwise make a full name search.
1293 try:
1294 getattr(self, cast(str, attr)) # Use cast to silence mypy.
1295 return True
1296 except (AttributeError, TypeError):
1297 return False
1298 # END handle membership
1300 def __getattr__(self, attr: str) -> T_IterableObj:
1301 attr = self._prefix + attr
1302 for item in self:
1303 if getattr(item, self._id_attr) == attr:
1304 return item
1305 # END for each item
1306 return list.__getattribute__(self, attr)
1308 def __getitem__( # type: ignore[override] # pyright: ignore[reportIncompatibleMethodOverride]
1309 self, index: Union[SupportsIndex, int, slice, str]
1310 ) -> T_IterableObj:
1311 if isinstance(index, int):
1312 return list.__getitem__(self, index)
1313 elif isinstance(index, slice):
1314 raise ValueError("Index should be an int or str")
1315 else:
1316 try:
1317 return getattr(self, cast(str, index))
1318 except AttributeError as e:
1319 raise IndexError(f"No item found with id {self._prefix}{index}") from e
1320 # END handle getattr
1322 def __delitem__(self, index: Union[SupportsIndex, int, slice, str]) -> None:
1323 delindex = cast(int, index)
1324 if isinstance(index, str):
1325 delindex = -1
1326 name = self._prefix + index
1327 for i, item in enumerate(self):
1328 if getattr(item, self._id_attr) == name:
1329 delindex = i
1330 break
1331 # END search index
1332 # END for each item
1333 if delindex == -1:
1334 raise IndexError("Item with name %s not found" % name)
1335 # END handle error
1336 # END get index to delete
1337 list.__delitem__(self, delindex)
1340@runtime_checkable
1341class IterableObj(Protocol):
1342 """Defines an interface for iterable items, so there is a uniform way to retrieve
1343 and iterate items within the git repository.
1345 Subclasses:
1347 * :class:`Submodule <git.objects.submodule.base.Submodule>`
1348 * :class:`Commit <git.objects.Commit>`
1349 * :class:`Reference <git.refs.reference.Reference>`
1350 * :class:`PushInfo <git.remote.PushInfo>`
1351 * :class:`FetchInfo <git.remote.FetchInfo>`
1352 * :class:`Remote <git.remote.Remote>`
1353 """
1355 __slots__ = ()
1357 _id_attribute_: str
1359 @classmethod
1360 @abstractmethod
1361 def iter_items(cls, repo: "Repo", *args: Any, **kwargs: Any) -> Iterator[T_IterableObj]:
1362 # Return-typed to be compatible with subtypes e.g. Remote.
1363 """Find (all) items of this type.
1365 Subclasses can specify `args` and `kwargs` differently, and may use them for
1366 filtering. However, when the method is called with no additional positional or
1367 keyword arguments, subclasses are obliged to to yield all items.
1369 :return:
1370 Iterator yielding Items
1371 """
1372 raise NotImplementedError("To be implemented by Subclass")
1374 @classmethod
1375 def list_items(cls, repo: "Repo", *args: Any, **kwargs: Any) -> IterableList[T_IterableObj]:
1376 """Find (all) items of this type and collect them into a list.
1378 For more information about the arguments, see :meth:`iter_items`.
1380 :note:
1381 Favor the :meth:`iter_items` method as it will avoid eagerly collecting all
1382 items. When there are many items, that can slow performance and increase
1383 memory usage.
1385 :return:
1386 list(Item,...) list of item instances
1387 """
1388 out_list: IterableList[T_IterableObj] = IterableList(cls._id_attribute_)
1389 out_list.extend(cls.iter_items(repo, *args, **kwargs))
1390 return out_list
1393class IterableClassWatcher(type):
1394 """Metaclass that issues :exc:`DeprecationWarning` when :class:`git.util.Iterable`
1395 is subclassed."""
1397 def __init__(cls, name: str, bases: Tuple[type, ...], clsdict: Dict[str, Any]) -> None:
1398 super().__init__(name, bases, clsdict)
1399 for base in bases:
1400 if type(base) is IterableClassWatcher:
1401 warnings.warn(
1402 f"GitPython Iterable subclassed by {name}."
1403 " Iterable is deprecated due to naming clash since v3.1.18"
1404 " and will be removed in 4.0.0."
1405 " Use IterableObj instead.",
1406 DeprecationWarning,
1407 stacklevel=2,
1408 )
1411class Iterable(metaclass=IterableClassWatcher):
1412 """Deprecated, use :class:`IterableObj` instead.
1414 Defines an interface for iterable items, so there is a uniform way to retrieve
1415 and iterate items within the git repository.
1416 """
1418 __slots__ = ()
1420 _id_attribute_ = "attribute that most suitably identifies your instance"
1422 @classmethod
1423 def iter_items(cls, repo: "Repo", *args: Any, **kwargs: Any) -> Any:
1424 """Deprecated, use :class:`IterableObj` instead.
1426 Find (all) items of this type.
1428 See :meth:`IterableObj.iter_items` for details on usage.
1430 :return:
1431 Iterator yielding Items
1432 """
1433 raise NotImplementedError("To be implemented by Subclass")
1435 @classmethod
1436 def list_items(cls, repo: "Repo", *args: Any, **kwargs: Any) -> Any:
1437 """Deprecated, use :class:`IterableObj` instead.
1439 Find (all) items of this type and collect them into a list.
1441 See :meth:`IterableObj.list_items` for details on usage.
1443 :return:
1444 list(Item,...) list of item instances
1445 """
1446 out_list: Any = IterableList(cls._id_attribute_)
1447 out_list.extend(cls.iter_items(repo, *args, **kwargs))
1448 return out_list
1451# } END classes