Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/git/util.py: 49%
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 TypeVar,
84 Union,
85 cast,
86 overload,
87)
89if TYPE_CHECKING:
90 from git.cmd import Git
91 from git.config import GitConfigParser, SectionConstraint
92 from git.remote import Remote
93 from git.repo.base import Repo
95from git.types import (
96 Files_TD,
97 Has_id_attribute,
98 HSH_TD,
99 Literal,
100 PathLike,
101 Protocol,
102 SupportsIndex,
103 Total_TD,
104 runtime_checkable,
105)
107# ---------------------------------------------------------------------
109T_IterableObj = TypeVar("T_IterableObj", bound=Union["IterableObj", "Has_id_attribute"], covariant=True)
110# So IterableList[Head] is subtype of IterableList[IterableObj].
112_logger = logging.getLogger(__name__)
115def _read_env_flag(name: str, default: bool) -> bool:
116 """Read a boolean flag from an environment variable.
118 :return:
119 The flag, or the `default` value if absent or ambiguous.
120 """
121 try:
122 value = os.environ[name]
123 except KeyError:
124 return default
126 _logger.warning(
127 "The %s environment variable is deprecated. Its effect has never been documented and changes without warning.",
128 name,
129 )
131 adjusted_value = value.strip().lower()
133 if adjusted_value in {"", "0", "false", "no"}:
134 return False
135 if adjusted_value in {"1", "true", "yes"}:
136 return True
137 _logger.warning("%s has unrecognized value %r, treating as %r.", name, value, default)
138 return default
141def _read_win_env_flag(name: str, default: bool) -> bool:
142 """Read a boolean flag from an environment variable on Windows.
144 :return:
145 On Windows, the flag, or the `default` value if absent or ambiguous.
146 On all other operating systems, ``False``.
148 :note:
149 This only accesses the environment on Windows.
150 """
151 return sys.platform == "win32" and _read_env_flag(name, default)
154#: We need an easy way to see if Appveyor TCs start failing,
155#: so the errors marked with this var are considered "acknowledged" ones, awaiting remedy,
156#: till then, we wish to hide them.
157HIDE_WINDOWS_KNOWN_ERRORS = _read_win_env_flag("HIDE_WINDOWS_KNOWN_ERRORS", True)
158HIDE_WINDOWS_FREEZE_ERRORS = _read_win_env_flag("HIDE_WINDOWS_FREEZE_ERRORS", True)
160# { Utility Methods
162T = TypeVar("T")
165def unbare_repo(func: Callable[..., T]) -> Callable[..., T]:
166 """Methods with this decorator raise :exc:`~git.exc.InvalidGitRepositoryError` if
167 they encounter a bare repository."""
169 from .exc import InvalidGitRepositoryError
171 @wraps(func)
172 def wrapper(self: "Remote", *args: Any, **kwargs: Any) -> T:
173 if self.repo.bare:
174 raise InvalidGitRepositoryError("Method '%s' cannot operate on bare repositories" % func.__name__)
175 # END bare method
176 return func(self, *args, **kwargs)
178 # END wrapper
180 return wrapper
183@contextlib.contextmanager
184def cwd(new_dir: PathLike) -> Generator[PathLike, None, None]:
185 """Context manager to temporarily change directory.
187 This is similar to :func:`contextlib.chdir` introduced in Python 3.11, but the
188 context manager object returned by a single call to this function is not reentrant.
189 """
190 old_dir = os.getcwd()
191 os.chdir(new_dir)
192 try:
193 yield new_dir
194 finally:
195 os.chdir(old_dir)
198@contextlib.contextmanager
199def patch_env(name: str, value: str) -> Generator[None, None, None]:
200 """Context manager to temporarily patch an environment variable."""
201 old_value = os.getenv(name)
202 os.environ[name] = value
203 try:
204 yield
205 finally:
206 if old_value is None:
207 del os.environ[name]
208 else:
209 os.environ[name] = old_value
212def rmtree(path: PathLike) -> None:
213 """Remove the given directory tree recursively.
215 :note:
216 We use :func:`shutil.rmtree` but adjust its behaviour to see whether files that
217 couldn't be deleted are read-only. Windows will not remove them in that case.
218 """
220 def handler(function: Callable[[str], Any], path: str, _excinfo: Any) -> None:
221 """Callback for :func:`shutil.rmtree`.
223 This works as either a ``onexc`` or ``onerror`` style callback.
224 """
225 # Is the error an access error?
226 os.chmod(path, stat.S_IWUSR)
228 try:
229 function(path)
230 except PermissionError as ex:
231 if HIDE_WINDOWS_KNOWN_ERRORS:
232 from unittest import SkipTest
234 raise SkipTest(f"FIXME: fails with: PermissionError\n {ex}") from ex
235 raise
237 if sys.platform != "win32":
238 shutil.rmtree(path)
239 elif sys.version_info >= (3, 12):
240 shutil.rmtree(path, onexc=handler)
241 else:
242 shutil.rmtree(path, onerror=handler)
245def rmfile(path: PathLike) -> None:
246 """Ensure file deleted also on *Windows* where read-only files need special
247 treatment."""
248 if osp.isfile(path):
249 if sys.platform == "win32":
250 os.chmod(path, 0o777)
251 os.remove(path)
254def stream_copy(source: BinaryIO, destination: BinaryIO, chunk_size: int = 512 * 1024) -> int:
255 """Copy all data from the `source` stream into the `destination` stream in chunks
256 of size `chunk_size`.
258 :return:
259 Number of bytes written
260 """
261 br = 0
262 while True:
263 chunk = source.read(chunk_size)
264 destination.write(chunk)
265 br += len(chunk)
266 if len(chunk) < chunk_size:
267 break
268 # END reading output stream
269 return br
272def join_path(a: PathLike, *p: PathLike) -> PathLike:
273 R"""Join path tokens together similar to osp.join, but always use ``/`` instead of
274 possibly ``\`` on Windows."""
275 path = os.fspath(a)
276 for b in p:
277 b = os.fspath(b)
278 if not b:
279 continue
280 if b.startswith("/"):
281 path += b[1:]
282 elif path == "" or path.endswith("/"):
283 path += b
284 else:
285 path += "/" + b
286 # END for each path token to add
287 return path
290if sys.platform == "win32":
292 def to_native_path_windows(path: PathLike) -> str:
293 path = os.fspath(path)
294 return path.replace("/", "\\")
296 def to_native_path_linux(path: PathLike) -> str:
297 path = os.fspath(path)
298 return path.replace("\\", "/")
300 to_native_path = to_native_path_windows
301else:
302 # No need for any work on Linux.
303 def to_native_path_linux(path: PathLike) -> str:
304 return os.fspath(path)
306 to_native_path = to_native_path_linux
309def join_path_native(a: PathLike, *p: PathLike) -> PathLike:
310 R"""Like :func:`join_path`, but makes sure an OS native path is returned.
312 This is only needed to play it safe on Windows and to ensure nice paths that only
313 use ``\``.
314 """
315 return to_native_path(join_path(a, *p))
318def assure_directory_exists(path: PathLike, is_file: bool = False) -> bool:
319 """Make sure that the directory pointed to by path exists.
321 :param is_file:
322 If ``True``, `path` is assumed to be a file and handled correctly.
323 Otherwise it must be a directory.
325 :return:
326 ``True`` if the directory was created, ``False`` if it already existed.
327 """
328 if is_file:
329 path = osp.dirname(path)
330 # END handle file
331 if not osp.isdir(path):
332 os.makedirs(path, exist_ok=True)
333 return True
334 return False
337def _get_exe_extensions() -> Sequence[str]:
338 PATHEXT = os.environ.get("PATHEXT", None)
339 if PATHEXT:
340 return tuple(p.upper() for p in PATHEXT.split(os.pathsep))
341 elif sys.platform == "win32":
342 return (".BAT", ".COM", ".EXE")
343 else:
344 return ()
347def py_where(program: str, path: Optional[PathLike] = None) -> List[str]:
348 """Perform a path search to assist :func:`is_cygwin_git`.
350 This is not robust for general use. It is an implementation detail of
351 :func:`is_cygwin_git`. When a search following all shell rules is needed,
352 :func:`shutil.which` can be used instead.
354 :note:
355 Neither this function nor :func:`shutil.which` will predict the effect of an
356 executable search on a native Windows system due to a :class:`subprocess.Popen`
357 call without ``shell=True``, because shell and non-shell executable search on
358 Windows differ considerably.
359 """
360 # From: http://stackoverflow.com/a/377028/548792
361 winprog_exts = _get_exe_extensions()
363 def is_exec(fpath: str) -> bool:
364 return (
365 osp.isfile(fpath)
366 and os.access(fpath, os.X_OK)
367 and (
368 sys.platform != "win32" or not winprog_exts or any(fpath.upper().endswith(ext) for ext in winprog_exts)
369 )
370 )
372 progs = []
373 if not path:
374 path = os.environ["PATH"]
375 for folder in os.fspath(path).split(os.pathsep):
376 folder = folder.strip('"')
377 if folder:
378 exe_path = osp.join(folder, program)
379 for f in [exe_path] + ["%s%s" % (exe_path, e) for e in winprog_exts]:
380 if is_exec(f):
381 progs.append(f)
382 return progs
385def _cygexpath(drive: Optional[str], path: str, expand_vars: bool = True) -> str:
386 if osp.isabs(path) and not drive:
387 # Invoked from `cygpath()` directly with `D:Apps\123`?
388 # It's an error, leave it alone just slashes)
389 p = path # convert to str if AnyPath given
390 else:
391 p = path and osp.normpath(osp.expandvars(osp.expanduser(path)) if expand_vars else path)
392 if osp.isabs(p):
393 if drive:
394 # Confusing, maybe a remote system should expand vars.
395 p = path
396 else:
397 p = cygpath(p)
398 elif drive:
399 p = "/proc/cygdrive/%s/%s" % (drive.lower(), p)
400 p_str = os.fspath(p) # ensure it is a str and not AnyPath
401 return p_str.replace("\\", "/")
404_cygpath_parsers: Tuple[Tuple[Pattern[str], Callable[..., str], bool], ...] = (
405 # See: https://msdn.microsoft.com/en-us/library/windows/desktop/aa365247(v=vs.85).aspx
406 # and: https://www.cygwin.com/cygwin-ug-net/using.html#unc-paths
407 (
408 re.compile(r"\\\\\?\\UNC\\([^\\]+)\\([^\\]+)(?:\\(.*))?"),
409 (lambda server, share, rest_path: "//%s/%s/%s" % (server, share, rest_path.replace("\\", "/"))),
410 False,
411 ),
412 (re.compile(r"\\\\\?\\(\w):[/\\](.*)"), (_cygexpath), False),
413 (re.compile(r"(\w):[/\\](.*)"), (_cygexpath), False),
414 (re.compile(r"file:(.*)", re.I), (lambda rest_path: rest_path), True),
415 (re.compile(r"(\w{2,}:.*)"), (lambda url: url), False), # remote URL, do nothing
416)
419def cygpath(path: str, expand_vars: bool = True) -> str:
420 """Use :meth:`git.cmd.Git.polish_url` instead, that works on any environment."""
421 path = os.fspath(path) # Ensure is str and not AnyPath.
422 # Fix to use Paths when 3.5 dropped. Or to be just str if only for URLs?
423 if not path.startswith(("/cygdrive", "//", "/proc/cygdrive")):
424 for regex, parser, recurse in _cygpath_parsers:
425 match = regex.match(path)
426 if match:
427 if parser is _cygexpath:
428 path = parser(*match.groups(), expand_vars=expand_vars)
429 else:
430 path = parser(*match.groups())
431 if recurse:
432 path = cygpath(path, expand_vars=expand_vars)
433 break
434 else:
435 path = _cygexpath(None, path, expand_vars=expand_vars)
437 return path
440_decygpath_regex = re.compile(r"(?:/proc)?/cygdrive/(\w)(/.*)?")
443def decygpath(path: PathLike) -> str:
444 path = os.fspath(path)
445 m = _decygpath_regex.match(path)
446 if m:
447 drive, rest_path = m.groups()
448 path = "%s:%s" % (drive.upper(), rest_path or "")
450 return path.replace("/", "\\")
453#: Store boolean flags denoting if a specific Git executable
454#: is from a Cygwin installation (since `cache_lru()` unsupported on PY2).
455_is_cygwin_cache: Dict[str, Optional[bool]] = {}
458def _is_cygwin_git(git_executable: str) -> bool:
459 is_cygwin = _is_cygwin_cache.get(git_executable) # type: Optional[bool]
460 if is_cygwin is None:
461 is_cygwin = False
462 try:
463 git_dir = osp.dirname(git_executable)
464 if not git_dir:
465 res = py_where(git_executable)
466 git_dir = osp.dirname(res[0]) if res else ""
468 # Just a name given, not a real path.
469 uname_cmd = osp.join(git_dir, "uname")
471 if not (Path(uname_cmd).is_file() and os.access(uname_cmd, os.X_OK)):
472 _logger.debug(f"Failed checking if running in CYGWIN: {uname_cmd} is not an executable")
473 _is_cygwin_cache[git_executable] = is_cygwin
474 return is_cygwin
476 process = subprocess.Popen([uname_cmd], stdout=subprocess.PIPE, universal_newlines=True)
477 uname_out, _ = process.communicate()
478 # retcode = process.poll()
479 is_cygwin = "CYGWIN" in uname_out
480 except Exception as ex:
481 _logger.debug("Failed checking if running in CYGWIN due to: %r", ex)
482 _is_cygwin_cache[git_executable] = is_cygwin
484 return is_cygwin
487@overload
488def is_cygwin_git(git_executable: None) -> Literal[False]: ...
491@overload
492def is_cygwin_git(git_executable: PathLike) -> bool: ...
495def is_cygwin_git(git_executable: Union[None, PathLike]) -> bool:
496 # TODO: when py3.7 support is dropped, use the new interpolation f"{variable=}"
497 _logger.debug(f"sys.platform={sys.platform!r}, git_executable={git_executable!r}")
498 if sys.platform != "cygwin":
499 return False
500 elif git_executable is None:
501 return False
502 else:
503 return _is_cygwin_git(str(git_executable))
506def get_user_id() -> str:
507 """:return: String identifying the currently active system user as ``name@node``"""
508 return "%s@%s" % (getpass.getuser(), platform.node())
511def finalize_process(proc: Union["subprocess.Popen[Any]", "Git.AutoInterrupt"], **kwargs: Any) -> None:
512 """Wait for the process (clone, fetch, pull or push) and handle its errors
513 accordingly."""
514 # TODO: No close proc-streams??
515 proc.wait(**kwargs)
518@overload
519def expand_path(p: None, expand_vars: bool = ...) -> None: ...
522@overload
523def expand_path(p: PathLike, expand_vars: bool = ...) -> Optional[PathLike]:
524 # TODO: Support for Python 3.5 has been dropped, so these overloads can be improved.
525 ...
528def expand_path(p: Union[None, PathLike], expand_vars: bool = True) -> Optional[PathLike]:
529 if p is None:
530 return None
531 try:
532 if isinstance(p, Path):
533 return p.resolve()
534 expanded_path = osp.expanduser(os.fspath(p))
535 if expand_vars:
536 expanded_path = osp.expandvars(expanded_path)
537 return osp.normpath(osp.abspath(expanded_path))
538 except Exception:
539 return None
542def remove_password_if_present(cmdline: Sequence[str]) -> List[str]:
543 """Parse any command line argument and if one of the elements is an URL with a
544 username and/or password, replace them by stars (in-place).
546 If nothing is found, this just returns the command line as-is.
548 This should be used for every log line that print a command line, as well as
549 exception messages.
550 """
551 new_cmdline = []
552 for index, to_parse in enumerate(cmdline):
553 new_cmdline.append(to_parse)
554 try:
555 url = urlsplit(to_parse)
556 # Remove password from the URL if present.
557 if url.password is None and url.username is None:
558 continue
560 if url.password is not None:
561 url = url._replace(netloc=url.netloc.replace(url.password, "*****"))
562 if url.username is not None:
563 url = url._replace(netloc=url.netloc.replace(url.username, "*****"))
564 new_cmdline[index] = urlunsplit(url)
565 except ValueError:
566 # This is not a valid URL.
567 continue
568 return new_cmdline
571# } END utilities
573# { Classes
576class RemoteProgress:
577 """Handler providing an interface to parse progress information emitted by
578 :manpage:`git-push(1)` and :manpage:`git-fetch(1)` and to dispatch callbacks
579 allowing subclasses to react to the progress."""
581 _num_op_codes: int = 9
582 (
583 BEGIN,
584 END,
585 COUNTING,
586 COMPRESSING,
587 WRITING,
588 RECEIVING,
589 RESOLVING,
590 FINDING_SOURCES,
591 CHECKING_OUT,
592 ) = [1 << x for x in range(_num_op_codes)]
593 STAGE_MASK = BEGIN | END
594 OP_MASK = ~STAGE_MASK
596 DONE_TOKEN = "done."
597 TOKEN_SEPARATOR = ", "
599 __slots__ = (
600 "_cur_line",
601 "_seen_ops",
602 "error_lines", # Lines that started with 'error:' or 'fatal:'.
603 "other_lines", # Lines not denoting progress (i.e.g. push-infos).
604 )
605 re_op_absolute = re.compile(r"(remote: )?([\w\s]+):\s+()(\d+)()(.*)")
606 re_op_relative = re.compile(r"(remote: )?([\w\s]+):\s+(\d+)% \((\d+)/(\d+)\)(.*)")
608 def __init__(self) -> None:
609 self._seen_ops: List[int] = []
610 self._cur_line: Optional[str] = None
611 self.error_lines: List[str] = []
612 self.other_lines: List[str] = []
614 def _parse_progress_line(self, line: AnyStr) -> None:
615 """Parse progress information from the given line as retrieved by
616 :manpage:`git-push(1)` or :manpage:`git-fetch(1)`.
618 - Lines that do not contain progress info are stored in :attr:`other_lines`.
619 - Lines that seem to contain an error (i.e. start with ``error:`` or ``fatal:``)
620 are stored in :attr:`error_lines`.
621 """
622 # handle
623 # Counting objects: 4, done.
624 # Compressing objects: 50% (1/2)
625 # Compressing objects: 100% (2/2)
626 # Compressing objects: 100% (2/2), done.
627 if isinstance(line, bytes): # mypy argues about ternary assignment.
628 line_str = line.decode("utf-8")
629 else:
630 line_str = line
631 self._cur_line = line_str
633 if self._cur_line.startswith(("error:", "fatal:")):
634 self.error_lines.append(self._cur_line)
635 return
637 cur_count, max_count = None, None
638 match = self.re_op_relative.match(line_str)
639 if match is None:
640 match = self.re_op_absolute.match(line_str)
642 if not match:
643 self.line_dropped(line_str)
644 self.other_lines.append(line_str)
645 return
646 # END could not get match
648 op_code = 0
649 _remote, op_name, _percent, cur_count, max_count, message = match.groups()
651 # Get operation ID.
652 if op_name == "Counting objects":
653 op_code |= self.COUNTING
654 elif op_name == "Compressing objects":
655 op_code |= self.COMPRESSING
656 elif op_name == "Writing objects":
657 op_code |= self.WRITING
658 elif op_name == "Receiving objects":
659 op_code |= self.RECEIVING
660 elif op_name == "Resolving deltas":
661 op_code |= self.RESOLVING
662 elif op_name == "Finding sources":
663 op_code |= self.FINDING_SOURCES
664 elif op_name == "Checking out files":
665 op_code |= self.CHECKING_OUT
666 else:
667 # Note: On Windows it can happen that partial lines are sent.
668 # Hence we get something like "CompreReceiving objects", which is
669 # a blend of "Compressing objects" and "Receiving objects".
670 # This can't really be prevented, so we drop the line verbosely
671 # to make sure we get informed in case the process spits out new
672 # commands at some point.
673 self.line_dropped(line_str)
674 # Note: Don't add this line to the other lines, as we have to silently
675 # drop it.
676 return
677 # END handle op code
679 # Figure out stage.
680 if op_code not in self._seen_ops:
681 self._seen_ops.append(op_code)
682 op_code |= self.BEGIN
683 # END begin opcode
685 if message is None:
686 message = ""
687 # END message handling
689 message = message.strip()
690 if message.endswith(self.DONE_TOKEN):
691 op_code |= self.END
692 message = message[: -len(self.DONE_TOKEN)]
693 # END end message handling
694 message = message.strip(self.TOKEN_SEPARATOR)
696 self.update(
697 op_code,
698 cur_count and float(cur_count),
699 max_count and float(max_count),
700 message,
701 )
703 def new_message_handler(self) -> Callable[[str], None]:
704 """
705 :return:
706 A progress handler suitable for :func:`~git.cmd.handle_process_output`,
707 passing lines on to this progress handler in a suitable format.
708 """
710 def handler(line: AnyStr) -> None:
711 return self._parse_progress_line(line.rstrip())
713 # END handler
715 return handler
717 def line_dropped(self, line: str) -> None:
718 """Called whenever a line could not be understood and was therefore dropped."""
719 pass
721 def update(
722 self,
723 op_code: int,
724 cur_count: Union[str, float],
725 max_count: Union[str, float, None] = None,
726 message: str = "",
727 ) -> None:
728 """Called whenever the progress changes.
730 :param op_code:
731 Integer allowing to be compared against Operation IDs and stage IDs.
733 Stage IDs are :const:`BEGIN` and :const:`END`. :const:`BEGIN` will only be
734 set once for each Operation ID as well as :const:`END`. It may be that
735 :const:`BEGIN` and :const:`END` are set at once in case only one progress
736 message was emitted due to the speed of the operation. Between
737 :const:`BEGIN` and :const:`END`, none of these flags will be set.
739 Operation IDs are all held within the :const:`OP_MASK`. Only one Operation
740 ID will be active per call.
742 :param cur_count:
743 Current absolute count of items.
745 :param max_count:
746 The maximum count of items we expect. It may be ``None`` in case there is no
747 maximum number of items or if it is (yet) unknown.
749 :param message:
750 In case of the :const:`WRITING` operation, it contains the amount of bytes
751 transferred. It may possibly be used for other purposes as well.
753 :note:
754 You may read the contents of the current line in
755 :attr:`self._cur_line <_cur_line>`.
756 """
757 pass
760class CallableRemoteProgress(RemoteProgress):
761 """A :class:`RemoteProgress` implementation forwarding updates to any callable.
763 :note:
764 Like direct instances of :class:`RemoteProgress`, instances of this
765 :class:`CallableRemoteProgress` class are not themselves directly callable.
766 Rather, instances of this class wrap a callable and forward to it. This should
767 therefore not be confused with :class:`git.types.CallableProgress`.
768 """
770 __slots__ = ("_callable",)
772 def __init__(self, fn: Callable[..., Any]) -> None:
773 self._callable = fn
774 super().__init__()
776 def update(self, *args: Any, **kwargs: Any) -> None:
777 self._callable(*args, **kwargs)
780class Actor:
781 """Actors hold information about a person acting on the repository. They can be
782 committers and authors or anything with a name and an email as mentioned in the git
783 log entries."""
785 # PRECOMPILED REGEX
786 name_only_regex = re.compile(r"<(.*)>")
787 name_email_regex = re.compile(r"(.*) <(.*?)>")
789 # ENVIRONMENT VARIABLES
790 # These are read when creating new commits.
791 env_author_name = "GIT_AUTHOR_NAME"
792 env_author_email = "GIT_AUTHOR_EMAIL"
793 env_committer_name = "GIT_COMMITTER_NAME"
794 env_committer_email = "GIT_COMMITTER_EMAIL"
796 # CONFIGURATION KEYS
797 conf_name = "name"
798 conf_email = "email"
800 __slots__ = ("name", "email")
802 def __init__(self, name: Optional[str], email: Optional[str]) -> None:
803 self.name = name
804 self.email = email
806 def __eq__(self, other: Any) -> bool:
807 return self.name == other.name and self.email == other.email
809 def __ne__(self, other: Any) -> bool:
810 return not (self == other)
812 def __hash__(self) -> int:
813 return hash((self.name, self.email))
815 def __str__(self) -> str:
816 return self.name if self.name else ""
818 def __repr__(self) -> str:
819 return '<git.Actor "%s <%s>">' % (self.name, self.email)
821 @classmethod
822 def _from_string(cls, string: str) -> "Actor":
823 """Create an :class:`Actor` from a string.
825 :param string:
826 The string, which is expected to be in regular git format::
828 John Doe <jdoe@example.com>
830 :return:
831 :class:`Actor`
832 """
833 m = cls.name_email_regex.search(string)
834 if m:
835 name, email = m.groups()
836 return Actor(name, email)
837 else:
838 m = cls.name_only_regex.search(string)
839 if m:
840 return Actor(m.group(1), None)
841 # Assume the best and use the whole string as name.
842 return Actor(string, None)
843 # END special case name
844 # END handle name/email matching
846 @classmethod
847 def _main_actor(
848 cls,
849 env_name: str,
850 env_email: str,
851 config_reader: Union[None, "GitConfigParser", "SectionConstraint[GitConfigParser]"] = None,
852 ) -> "Actor":
853 actor = Actor("", "")
854 user_id = None # We use this to avoid multiple calls to getpass.getuser().
856 def default_email() -> str:
857 nonlocal user_id
858 if not user_id:
859 user_id = get_user_id()
860 return user_id
862 def default_name() -> str:
863 return default_email().split("@")[0]
865 for attr, evar, cvar, default in (
866 ("name", env_name, cls.conf_name, default_name),
867 ("email", env_email, cls.conf_email, default_email),
868 ):
869 try:
870 val = os.environ[evar]
871 setattr(actor, attr, val)
872 except KeyError:
873 if config_reader is not None:
874 try:
875 val = config_reader.get("user", cvar)
876 except Exception:
877 val = default()
878 setattr(actor, attr, val)
879 # END config-reader handling
880 if not getattr(actor, attr):
881 setattr(actor, attr, default())
882 # END handle name
883 # END for each item to retrieve
884 return actor
886 @classmethod
887 def committer(
888 cls, config_reader: Union[None, "GitConfigParser", "SectionConstraint[GitConfigParser]"] = None
889 ) -> "Actor":
890 """
891 :return:
892 :class:`Actor` instance corresponding to the configured committer. It
893 behaves similar to the git implementation, such that the environment will
894 override configuration values of `config_reader`. If no value is set at all,
895 it will be generated.
897 :param config_reader:
898 ConfigReader to use to retrieve the values from in case they are not set in
899 the environment.
900 """
901 return cls._main_actor(cls.env_committer_name, cls.env_committer_email, config_reader)
903 @classmethod
904 def author(
905 cls, config_reader: Union[None, "GitConfigParser", "SectionConstraint[GitConfigParser]"] = None
906 ) -> "Actor":
907 """Same as :meth:`committer`, but defines the main author. It may be specified
908 in the environment, but defaults to the committer."""
909 return cls._main_actor(cls.env_author_name, cls.env_author_email, config_reader)
912class Stats:
913 """Represents stat information as presented by git at the end of a merge. It is
914 created from the output of a diff operation.
916 Example::
918 c = Commit( sha1 )
919 s = c.stats
920 s.total # full-stat-dict
921 s.files # dict( filepath : stat-dict )
923 ``stat-dict``
925 A dictionary with the following keys and values::
927 deletions = number of deleted lines as int
928 insertions = number of inserted lines as int
929 lines = total number of lines changed as int, or deletions + insertions
930 change_type = type of change as str, A|C|D|M|R|T|U|X|B
932 ``full-stat-dict``
934 In addition to the items in the stat-dict, it features additional information::
936 files = number of changed files as int
937 """
939 __slots__ = ("total", "files")
941 def __init__(self, total: Total_TD, files: Dict[PathLike, Files_TD]) -> None:
942 self.total = total
943 self.files = files
945 @classmethod
946 def _list_from_string(cls, repo: "Repo", text: str) -> "Stats":
947 """Create a :class:`Stats` object from output retrieved by
948 :manpage:`git-diff(1)`.
950 :return:
951 :class:`git.Stats`
952 """
954 hsh: HSH_TD = {
955 "total": {"insertions": 0, "deletions": 0, "lines": 0, "files": 0},
956 "files": {},
957 }
958 for line in text.splitlines():
959 (change_type, raw_insertions, raw_deletions, filename) = line.split("\t")
960 insertions = raw_insertions != "-" and int(raw_insertions) or 0
961 deletions = raw_deletions != "-" and int(raw_deletions) or 0
962 hsh["total"]["insertions"] += insertions
963 hsh["total"]["deletions"] += deletions
964 hsh["total"]["lines"] += insertions + deletions
965 hsh["total"]["files"] += 1
966 files_dict: Files_TD = {
967 "insertions": insertions,
968 "deletions": deletions,
969 "lines": insertions + deletions,
970 "change_type": change_type,
971 }
972 hsh["files"][filename.strip()] = files_dict
973 return Stats(hsh["total"], hsh["files"])
976class IndexFileSHA1Writer:
977 """Wrapper around a file-like object that remembers the SHA1 of the data written to
978 it. It will write a sha when the stream is closed or if asked for explicitly using
979 :meth:`write_sha`.
981 Only useful to the index file.
983 :note:
984 Based on the dulwich project.
985 """
987 __slots__ = ("f", "sha1")
989 def __init__(self, f: IO[bytes]) -> None:
990 self.f = f
991 self.sha1 = make_sha(b"")
993 def write(self, data: bytes) -> int:
994 self.sha1.update(data)
995 return self.f.write(data)
997 def write_sha(self) -> bytes:
998 sha = self.sha1.digest()
999 self.f.write(sha)
1000 return sha
1002 def close(self) -> bytes:
1003 sha = self.write_sha()
1004 self.f.close()
1005 return sha
1007 def tell(self) -> int:
1008 return self.f.tell()
1011class LockFile:
1012 """Provides methods to obtain, check for, and release a file based lock which
1013 should be used to handle concurrent access to the same file.
1015 As we are a utility class to be derived from, we only use protected methods.
1017 Locks will automatically be released on destruction.
1018 """
1020 __slots__ = ("_file_path", "_owns_lock")
1022 def __init__(self, file_path: PathLike) -> None:
1023 self._file_path = file_path
1024 self._owns_lock = False
1026 def __del__(self) -> None:
1027 self._release_lock()
1029 def _lock_file_path(self) -> str:
1030 """:return: Path to lockfile"""
1031 return "%s.lock" % (self._file_path)
1033 def _has_lock(self) -> bool:
1034 """
1035 :return:
1036 True if we have a lock and if the lockfile still exists
1038 :raise AssertionError:
1039 If our lock-file does not exist.
1040 """
1041 return self._owns_lock
1043 def _obtain_lock_or_raise(self) -> None:
1044 """Create a lock file as flag for other instances, mark our instance as
1045 lock-holder.
1047 :raise IOError:
1048 If a lock was already present or a lock file could not be written.
1049 """
1050 if self._has_lock():
1051 return
1052 lock_file = self._lock_file_path()
1053 if osp.isfile(lock_file):
1054 raise IOError(
1055 "Lock for file %r did already exist, delete %r in case the lock is illegal"
1056 % (self._file_path, lock_file)
1057 )
1059 try:
1060 with open(lock_file, mode="w"):
1061 pass
1062 except OSError as e:
1063 raise IOError(str(e)) from e
1065 self._owns_lock = True
1067 def _obtain_lock(self) -> None:
1068 """The default implementation will raise if a lock cannot be obtained.
1070 Subclasses may override this method to provide a different implementation.
1071 """
1072 return self._obtain_lock_or_raise()
1074 def _release_lock(self) -> None:
1075 """Release our lock if we have one."""
1076 if not self._has_lock():
1077 return
1079 # If someone removed our file beforehand, lets just flag this issue instead of
1080 # failing, to make it more usable.
1081 lfp = self._lock_file_path()
1082 try:
1083 rmfile(lfp)
1084 except OSError:
1085 pass
1086 self._owns_lock = False
1089class BlockingLockFile(LockFile):
1090 """The lock file will block until a lock could be obtained, or fail after a
1091 specified timeout.
1093 :note:
1094 If the directory containing the lock was removed, an exception will be raised
1095 during the blocking period, preventing hangs as the lock can never be obtained.
1096 """
1098 __slots__ = ("_check_interval", "_max_block_time")
1100 def __init__(
1101 self,
1102 file_path: PathLike,
1103 check_interval_s: float = 0.3,
1104 max_block_time_s: int = sys.maxsize,
1105 ) -> None:
1106 """Configure the instance.
1108 :param check_interval_s:
1109 Period of time to sleep until the lock is checked the next time.
1110 By default, it waits a nearly unlimited time.
1112 :param max_block_time_s:
1113 Maximum amount of seconds we may lock.
1114 """
1115 super().__init__(file_path)
1116 self._check_interval = check_interval_s
1117 self._max_block_time = max_block_time_s
1119 def _obtain_lock(self) -> None:
1120 """This method blocks until it obtained the lock, or raises :exc:`IOError` if it
1121 ran out of time or if the parent directory was not available anymore.
1123 If this method returns, you are guaranteed to own the lock.
1124 """
1125 starttime = time.time()
1126 maxtime = starttime + float(self._max_block_time)
1127 while True:
1128 try:
1129 super()._obtain_lock()
1130 except IOError as e:
1131 # synity check: if the directory leading to the lockfile is not
1132 # readable anymore, raise an exception
1133 curtime = time.time()
1134 if not osp.isdir(osp.dirname(self._lock_file_path())):
1135 msg = "Directory containing the lockfile %r was not readable anymore after waiting %g seconds" % (
1136 self._lock_file_path(),
1137 curtime - starttime,
1138 )
1139 raise IOError(msg) from e
1140 # END handle missing directory
1142 if curtime >= maxtime:
1143 msg = "Waited %g seconds for lock at %r" % (
1144 maxtime - starttime,
1145 self._lock_file_path(),
1146 )
1147 raise IOError(msg) from e
1148 # END abort if we wait too long
1149 time.sleep(self._check_interval)
1150 else:
1151 break
1152 # END endless loop
1155class IterableList(List[T_IterableObj]): # type: ignore[type-var]
1156 """List of iterable objects allowing to query an object by id or by named index::
1158 heads = repo.heads
1159 heads.master
1160 heads['master']
1161 heads[0]
1163 Iterable parent objects:
1165 * :class:`Commit <git.objects.Commit>`
1166 * :class:`Submodule <git.objects.submodule.base.Submodule>`
1167 * :class:`Reference <git.refs.reference.Reference>`
1168 * :class:`FetchInfo <git.remote.FetchInfo>`
1169 * :class:`PushInfo <git.remote.PushInfo>`
1171 Iterable via inheritance:
1173 * :class:`Head <git.refs.head.Head>`
1174 * :class:`TagReference <git.refs.tag.TagReference>`
1175 * :class:`RemoteReference <git.refs.remote.RemoteReference>`
1177 This requires an ``id_attribute`` name to be set which will be queried from its
1178 contained items to have a means for comparison.
1180 A prefix can be specified which is to be used in case the id returned by the items
1181 always contains a prefix that does not matter to the user, so it can be left out.
1182 """
1184 __slots__ = ("_id_attr", "_prefix")
1186 def __new__(cls, id_attr: str, prefix: str = "") -> "IterableList[T_IterableObj]":
1187 return super().__new__(cls)
1189 def __init__(self, id_attr: str, prefix: str = "") -> None:
1190 super().__init__()
1191 self._id_attr = id_attr
1192 self._prefix = prefix
1194 def __contains__(self, attr: object) -> bool:
1195 # First try identity match for performance.
1196 try:
1197 rval = list.__contains__(self, attr)
1198 if rval:
1199 return rval
1200 except (AttributeError, TypeError):
1201 pass
1202 # END handle match
1204 # Otherwise make a full name search.
1205 try:
1206 getattr(self, cast(str, attr)) # Use cast to silence mypy.
1207 return True
1208 except (AttributeError, TypeError):
1209 return False
1210 # END handle membership
1212 def __getattr__(self, attr: str) -> T_IterableObj:
1213 attr = self._prefix + attr
1214 for item in self:
1215 if getattr(item, self._id_attr) == attr:
1216 return item
1217 # END for each item
1218 return list.__getattribute__(self, attr)
1220 def __getitem__( # type: ignore[override] # pyright: ignore[reportIncompatibleMethodOverride]
1221 self, index: Union[SupportsIndex, int, slice, str]
1222 ) -> T_IterableObj:
1223 if isinstance(index, int):
1224 return list.__getitem__(self, index)
1225 elif isinstance(index, slice):
1226 raise ValueError("Index should be an int or str")
1227 else:
1228 try:
1229 return getattr(self, cast(str, index))
1230 except AttributeError as e:
1231 raise IndexError(f"No item found with id {self._prefix}{index}") from e
1232 # END handle getattr
1234 def __delitem__(self, index: Union[SupportsIndex, int, slice, str]) -> None:
1235 delindex = cast(int, index)
1236 if isinstance(index, str):
1237 delindex = -1
1238 name = self._prefix + index
1239 for i, item in enumerate(self):
1240 if getattr(item, self._id_attr) == name:
1241 delindex = i
1242 break
1243 # END search index
1244 # END for each item
1245 if delindex == -1:
1246 raise IndexError("Item with name %s not found" % name)
1247 # END handle error
1248 # END get index to delete
1249 list.__delitem__(self, delindex)
1252@runtime_checkable
1253class IterableObj(Protocol):
1254 """Defines an interface for iterable items, so there is a uniform way to retrieve
1255 and iterate items within the git repository.
1257 Subclasses:
1259 * :class:`Submodule <git.objects.submodule.base.Submodule>`
1260 * :class:`Commit <git.objects.Commit>`
1261 * :class:`Reference <git.refs.reference.Reference>`
1262 * :class:`PushInfo <git.remote.PushInfo>`
1263 * :class:`FetchInfo <git.remote.FetchInfo>`
1264 * :class:`Remote <git.remote.Remote>`
1265 """
1267 __slots__ = ()
1269 _id_attribute_: str
1271 @classmethod
1272 @abstractmethod
1273 def iter_items(cls, repo: "Repo", *args: Any, **kwargs: Any) -> Iterator[T_IterableObj]:
1274 # Return-typed to be compatible with subtypes e.g. Remote.
1275 """Find (all) items of this type.
1277 Subclasses can specify `args` and `kwargs` differently, and may use them for
1278 filtering. However, when the method is called with no additional positional or
1279 keyword arguments, subclasses are obliged to to yield all items.
1281 :return:
1282 Iterator yielding Items
1283 """
1284 raise NotImplementedError("To be implemented by Subclass")
1286 @classmethod
1287 def list_items(cls, repo: "Repo", *args: Any, **kwargs: Any) -> IterableList[T_IterableObj]:
1288 """Find (all) items of this type and collect them into a list.
1290 For more information about the arguments, see :meth:`iter_items`.
1292 :note:
1293 Favor the :meth:`iter_items` method as it will avoid eagerly collecting all
1294 items. When there are many items, that can slow performance and increase
1295 memory usage.
1297 :return:
1298 list(Item,...) list of item instances
1299 """
1300 out_list: IterableList[T_IterableObj] = IterableList(cls._id_attribute_)
1301 out_list.extend(cls.iter_items(repo, *args, **kwargs))
1302 return out_list
1305class IterableClassWatcher(type):
1306 """Metaclass that issues :exc:`DeprecationWarning` when :class:`git.util.Iterable`
1307 is subclassed."""
1309 def __init__(cls, name: str, bases: Tuple[type, ...], clsdict: Dict[str, Any]) -> None:
1310 super().__init__(name, bases, clsdict)
1311 for base in bases:
1312 if type(base) is IterableClassWatcher:
1313 warnings.warn(
1314 f"GitPython Iterable subclassed by {name}."
1315 " Iterable is deprecated due to naming clash since v3.1.18"
1316 " and will be removed in 4.0.0."
1317 " Use IterableObj instead.",
1318 DeprecationWarning,
1319 stacklevel=2,
1320 )
1323class Iterable(metaclass=IterableClassWatcher):
1324 """Deprecated, use :class:`IterableObj` instead.
1326 Defines an interface for iterable items, so there is a uniform way to retrieve
1327 and iterate items within the git repository.
1328 """
1330 __slots__ = ()
1332 _id_attribute_ = "attribute that most suitably identifies your instance"
1334 @classmethod
1335 def iter_items(cls, repo: "Repo", *args: Any, **kwargs: Any) -> Any:
1336 """Deprecated, use :class:`IterableObj` instead.
1338 Find (all) items of this type.
1340 See :meth:`IterableObj.iter_items` for details on usage.
1342 :return:
1343 Iterator yielding Items
1344 """
1345 raise NotImplementedError("To be implemented by Subclass")
1347 @classmethod
1348 def list_items(cls, repo: "Repo", *args: Any, **kwargs: Any) -> Any:
1349 """Deprecated, use :class:`IterableObj` instead.
1351 Find (all) items of this type and collect them into a list.
1353 See :meth:`IterableObj.list_items` for details on usage.
1355 :return:
1356 list(Item,...) list of item instances
1357 """
1358 out_list: Any = IterableList(cls._id_attribute_)
1359 out_list.extend(cls.iter_items(repo, *args, **kwargs))
1360 return out_list
1363# } END classes