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")
32import contextlib
33import getpass
34import logging
35import ntpath
36import os
37import os.path as osp
38import platform
39import re
40import shutil
41import stat
42import subprocess
43import time
44import warnings
45from abc import abstractmethod
46from functools import wraps
47from pathlib import Path
49# typing ---------------------------------------------------------
50from typing import (
51 IO,
52 TYPE_CHECKING,
53 Any,
54 AnyStr,
55 Callable,
56 Dict,
57 Generator,
58 Iterator,
59 List,
60 Optional,
61 Pattern,
62 Sequence,
63 Tuple,
64 Type,
65 TypeVar,
66 Union,
67 cast,
68 overload,
69)
70from urllib.parse import urlsplit, urlunsplit
72# NOTE: Unused imports can be improved now that CI testing has fully resumed. Some of
73# these be used indirectly through other GitPython modules, which avoids having to write
74# gitdb all the time in their imports. They are not in __all__, at least currently,
75# because they could be removed or changed at any time, and so should not be considered
76# conceptually public to code outside GitPython. Linters of course do not like it.
77from gitdb.util import (
78 LazyMixin, # noqa: F401
79 LockedFD, # noqa: F401
80 bin_to_hex, # noqa: F401
81 file_contents_ro, # noqa: F401
82 file_contents_ro_filepath, # noqa: F401
83 hex_to_bin, # noqa: F401
84 make_sha,
85 to_bin_sha, # noqa: F401
86 to_hex_sha, # noqa: F401
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 HSH_TD,
97 Files_TD,
98 Has_id_attribute,
99 Literal,
100 PathLike,
101 Protocol,
102 SupportsIndex,
103 SupportsRead,
104 SupportsWrite,
105 Total_TD,
106 runtime_checkable,
107)
109# ---------------------------------------------------------------------
111T_IterableObj = TypeVar("T_IterableObj", bound=Union["IterableObj", "Has_id_attribute"], covariant=True)
112# So IterableList[Head] is subtype of IterableList[IterableObj].
113T_Actor = TypeVar("T_Actor", bound="Actor")
115_logger = logging.getLogger(__name__)
118def _read_env_flag(name: str, default: bool) -> bool:
119 """Read a boolean flag from an environment variable.
121 :return:
122 The flag, or the `default` value if absent or ambiguous.
123 """
124 try:
125 value = os.environ[name]
126 except KeyError:
127 return default
129 _logger.warning(
130 "The %s environment variable is deprecated. Its effect has never been documented and changes without warning.",
131 name,
132 )
134 adjusted_value = value.strip().lower()
136 if adjusted_value in {"", "0", "false", "no"}:
137 return False
138 if adjusted_value in {"1", "true", "yes"}:
139 return True
140 _logger.warning("%s has unrecognized value %r, treating as %r.", name, value, default)
141 return default
144def _read_win_env_flag(name: str, default: bool) -> bool:
145 """Read a boolean flag from an environment variable on Windows.
147 :return:
148 On Windows, the flag, or the `default` value if absent or ambiguous.
149 On all other operating systems, ``False``.
151 :note:
152 This only accesses the environment on Windows.
153 """
154 return sys.platform == "win32" and _read_env_flag(name, default)
157#: We need an easy way to see if Appveyor TCs start failing,
158#: so the errors marked with this var are considered "acknowledged" ones, awaiting remedy,
159#: till then, we wish to hide them.
160HIDE_WINDOWS_KNOWN_ERRORS = _read_win_env_flag("HIDE_WINDOWS_KNOWN_ERRORS", True)
161HIDE_WINDOWS_FREEZE_ERRORS = _read_win_env_flag("HIDE_WINDOWS_FREEZE_ERRORS", True)
163# { Utility Methods
165T = TypeVar("T")
168def unbare_repo(func: Callable[..., T]) -> Callable[..., T]:
169 """Methods with this decorator raise :exc:`~git.exc.InvalidGitRepositoryError` if
170 they encounter a bare repository."""
172 from .exc import InvalidGitRepositoryError
174 @wraps(func)
175 def wrapper(self: "Remote", *args: Any, **kwargs: Any) -> T:
176 if self.repo.bare:
177 raise InvalidGitRepositoryError("Method '%s' cannot operate on bare repositories" % func.__name__)
178 # END bare method
179 return func(self, *args, **kwargs)
181 # END wrapper
183 return wrapper
186@contextlib.contextmanager
187def cwd(new_dir: PathLike) -> Generator[PathLike, None, None]:
188 """Context manager to temporarily change directory.
190 This is similar to :func:`contextlib.chdir` introduced in Python 3.11, but the
191 context manager object returned by a single call to this function is not reentrant.
192 """
193 old_dir = os.getcwd()
194 os.chdir(new_dir)
195 try:
196 yield new_dir
197 finally:
198 os.chdir(old_dir)
201@contextlib.contextmanager
202def patch_env(name: str, value: str) -> Generator[None, None, None]:
203 """Context manager to temporarily patch an environment variable."""
204 old_value = os.getenv(name)
205 os.environ[name] = value
206 try:
207 yield
208 finally:
209 if old_value is None:
210 del os.environ[name]
211 else:
212 os.environ[name] = old_value
215def rmtree(path: PathLike) -> None:
216 """Remove the given directory tree recursively.
218 :note:
219 We use :func:`shutil.rmtree` but adjust its behaviour to see whether files that
220 couldn't be deleted are read-only. Windows will not remove them in that case.
221 """
223 def handler(function: Callable[[str], Any], path: str, _excinfo: Any) -> None:
224 """Callback for :func:`shutil.rmtree`.
226 This works as either a ``onexc`` or ``onerror`` style callback.
227 """
228 # Is the error an access error?
229 os.chmod(path, stat.S_IWUSR)
231 try:
232 function(path)
233 except PermissionError as ex:
234 if HIDE_WINDOWS_KNOWN_ERRORS:
235 from unittest import SkipTest
237 raise SkipTest(f"FIXME: fails with: PermissionError\n {ex}") from ex
238 raise
240 if sys.platform != "win32":
241 shutil.rmtree(path)
242 elif sys.version_info >= (3, 12):
243 shutil.rmtree(path, onexc=handler)
244 else:
245 shutil.rmtree(path, onerror=handler)
248def rmfile(path: PathLike) -> None:
249 """Ensure file deleted also on *Windows* where read-only files need special
250 treatment."""
251 if osp.isfile(path):
252 if sys.platform == "win32":
253 os.chmod(path, 0o777)
254 os.remove(path)
257def stream_copy(source: SupportsRead[AnyStr], destination: SupportsWrite[AnyStr], chunk_size: int = 512 * 1024) -> int:
258 """Copy all data from the `source` stream into the `destination` stream in chunks
259 of size `chunk_size`.
261 :return:
262 Number of bytes written
263 """
264 br = 0
265 while True:
266 chunk = source.read(chunk_size)
267 destination.write(chunk)
268 br += len(chunk)
269 if len(chunk) < chunk_size:
270 break
271 # END reading output stream
272 return br
275def join_path(a: PathLike, *p: PathLike) -> PathLike:
276 R"""Join path tokens together similar to osp.join, but always use ``/`` instead of
277 possibly ``\`` on Windows."""
278 path = os.fspath(a)
279 for b in p:
280 b = os.fspath(b)
281 if not b:
282 continue
283 if b.startswith("/"):
284 path += b[1:]
285 elif path == "" or path.endswith("/"):
286 path += b
287 else:
288 path += "/" + b
289 # END for each path token to add
290 return path
293if sys.platform == "win32":
295 def to_native_path_windows(path: PathLike) -> str:
296 path = os.fspath(path)
297 return path.replace("/", "\\")
299 def to_native_path_linux(path: PathLike) -> str:
300 path = os.fspath(path)
301 return path.replace("\\", "/")
303 to_native_path = to_native_path_windows
304else:
305 # No need for any work on Linux.
306 def to_native_path_linux(path: PathLike) -> str:
307 return os.fspath(path)
309 to_native_path = to_native_path_linux
312def join_path_native(a: PathLike, *p: PathLike) -> PathLike:
313 R"""Like :func:`join_path`, but makes sure an OS native path is returned.
315 This is only needed to play it safe on Windows and to ensure nice paths that only
316 use ``\``.
317 """
318 return to_native_path(join_path(a, *p))
321def _is_path_rooted(path: PathLike) -> bool:
322 r"""Whether ``path`` has a root, including one encoded in a UNC drive.
324 On Windows, ``\directory`` is rooted on the current drive without being
325 absolute, while ``C:\directory`` has both a drive and a root. In contrast,
326 ``directory`` and the drive-relative ``C:directory`` have no root.
327 UNC paths are rooted: ``\\server\share`` stores the share in the drive
328 returned by :func:`os.path.splitdrive`, while ``\\server\share\directory``
329 additionally has a rooted tail.
330 On POSIX, which has no drive concept, this simply distinguishes absolute
331 paths such as ``/directory`` from relative paths such as ``directory``.
332 """
333 drive, tail = osp.splitdrive(os.fspath(path))
334 separators = (os.sep,) if os.altsep is None else (os.sep, os.altsep)
335 return tail.startswith(separators) or drive.startswith(separators)
338def _to_relative_path(root: PathLike, path: PathLike) -> str:
339 r"""Return a normalized Git-style path confined to ``root``.
341 A Windows path such as ``\directory`` is rooted but not absolute. Resolve it
342 against the drive of ``root`` rather than treating it as relative to ``root``.
343 Drive-relative paths such as ``C:directory`` are rejected because their meaning
344 depends on process-global per-drive state.
346 For example, with ``root`` set to ``C:\repo`` on Windows:
348 * ``directory\file`` -> ``directory/file``
349 * ``directory\`` -> ``directory/``
350 * ``C:\repo\directory\file`` -> ``directory/file``
351 * ``\repo\directory\file`` -> ``directory/file``
352 * ``C:directory\file`` -> :exc:`ValueError`
353 * ``C:\other\file`` -> :exc:`ValueError`
355 On POSIX, ``/repo/directory/file`` under ``/repo`` similarly becomes
356 ``directory/file``. A trailing separator is preserved as a Git-style ``/``.
357 """
358 path_str = os.fspath(path)
359 if not path_str:
360 return path_str
362 drive, _tail = osp.splitdrive(path_str)
363 rooted = _is_path_rooted(path_str)
364 if drive and not rooted:
365 raise ValueError("Drive-relative path %r is not supported" % path_str)
367 root_abs = osp.abspath(os.fspath(root))
368 path_abs = osp.abspath(osp.join(root_abs, path_str))
369 try:
370 common_path = osp.commonpath([root_abs, path_abs])
371 except ValueError as e:
372 raise ValueError("Path %r is not in repository at %r" % (path_str, root_abs)) from e
373 if common_path != root_abs:
374 raise ValueError("Path %r is not in repository at %r" % (path_str, root_abs))
376 relative_path = to_native_path_linux(osp.relpath(path_abs, root_abs))
377 separators = (os.sep,) if os.altsep is None else (os.sep, os.altsep)
378 if path_str.endswith(separators) and relative_path != "." and not relative_path.endswith("/"):
379 relative_path += "/"
380 return relative_path
383_HFS_IGNORABLES = str.maketrans(
384 "", "", "\u200c\u200d\u200e\u200f\u202a\u202b\u202c\u202d\u202e\u206a\u206b\u206c\u206d\u206e\u206f\ufeff"
385)
387# Match Git's is_ntfs_dotgitmodules in path.c on a lowercased, trimmed name.
388# Besides gitmod~1..4, fallback aliases have exactly eight ASCII characters:
389# a shrinking prefix of "gi7eba", "~", and digits with no leading zero.
390# Explicit digit counts avoid accepting shorter/longer names or Unicode digits.
391_NTFS_DOTGITMODULES_SHORT_NAME = re.compile(
392 r"(?:gitmod~[1-4]|gi7eba~[1-9]|gi7eb~[1-9][0-9]|gi7e~[1-9][0-9]{2}|"
393 r"gi7~[1-9][0-9]{3}|gi~[1-9][0-9]{4}|g~[1-9][0-9]{5}|~[1-9][0-9]{6})"
394)
397def _validate_repo_path(path: PathLike, mode: Union[int, None] = None) -> None:
398 """Reject unsafe tree/index paths without normalizing away their components.
400 Protect Git metadata aliases on NTFS and HFS even when writing on another
401 platform. Other POSIX filename characters, including newlines, remain valid.
403 :param mode:
404 Mode of the index or tree entry the path belongs to, where one is known.
405 Git refuses a symbolic link that aliases ``.gitmodules``, since the
406 submodule configuration would then be read through the link, so that
407 name is only rejected once the mode says the entry is a link.
408 """
409 name = os.fspath(path)
410 if not name or "\0" in name or ntpath.splitdrive(name)[0] or name.startswith("/"):
411 raise ValueError("Invalid repository path %r" % name)
412 if os.name == "nt" and "\\" in name:
413 raise ValueError("Index paths must use '/' separators: %r" % name)
414 for component in name.split("/"):
415 if component in ("", ".", ".."):
416 raise ValueError("Invalid repository path %r" % name)
417 # NTFS recognizes backslashes and alternate data streams in metadata names.
418 for part in component.split("\\"):
419 ntfs_name = part.split(":", 1)[0].rstrip(" .").lower()
420 hfs_name = part.translate(_HFS_IGNORABLES).lower()
421 if ntfs_name in (".git", "git~1") or hfs_name == ".git":
422 raise ValueError("Repository path aliases Git metadata: %r" % name)
423 if mode is not None and stat.S_ISLNK(mode):
424 if (
425 ntfs_name == ".gitmodules"
426 or hfs_name == ".gitmodules"
427 or _NTFS_DOTGITMODULES_SHORT_NAME.fullmatch(ntfs_name) is not None
428 ):
429 raise ValueError("Symbolic link aliases the submodule configuration: %r" % name)
432def assure_directory_exists(path: PathLike, is_file: bool = False) -> bool:
433 """Make sure that the directory pointed to by path exists.
435 :param is_file:
436 If ``True``, `path` is assumed to be a file and handled correctly.
437 Otherwise it must be a directory.
439 :return:
440 ``True`` if the directory was created, ``False`` if it already existed.
441 """
442 if is_file:
443 path = osp.dirname(path)
444 # END handle file
445 if not osp.isdir(path):
446 os.makedirs(path, exist_ok=True)
447 return True
448 return False
451def _get_exe_extensions() -> Sequence[str]:
452 PATHEXT = os.environ.get("PATHEXT", None)
453 if PATHEXT:
454 return tuple(p.upper() for p in PATHEXT.split(os.pathsep))
455 elif sys.platform == "win32":
456 return (".BAT", ".COM", ".EXE")
457 else:
458 return ()
461def py_where(program: str, path: Optional[PathLike] = None) -> List[str]:
462 """Perform a path search to assist :func:`is_cygwin_git`.
464 This is not robust for general use. It is an implementation detail of
465 :func:`is_cygwin_git`. When a search following all shell rules is needed,
466 :func:`shutil.which` can be used instead.
468 :note:
469 Neither this function nor :func:`shutil.which` will predict the effect of an
470 executable search on a native Windows system due to a :class:`subprocess.Popen`
471 call without ``shell=True``, because shell and non-shell executable search on
472 Windows differ considerably.
473 """
474 # From: http://stackoverflow.com/a/377028/548792
475 winprog_exts = _get_exe_extensions()
477 def is_exec(fpath: str) -> bool:
478 return (
479 osp.isfile(fpath)
480 and os.access(fpath, os.X_OK)
481 and (
482 sys.platform != "win32" or not winprog_exts or any(fpath.upper().endswith(ext) for ext in winprog_exts)
483 )
484 )
486 progs = []
487 if not path:
488 path = os.environ["PATH"]
489 for folder in os.fspath(path).split(os.pathsep):
490 folder = folder.strip('"')
491 if folder:
492 exe_path = osp.join(folder, program)
493 for f in [exe_path] + ["%s%s" % (exe_path, e) for e in winprog_exts]:
494 if is_exec(f):
495 progs.append(f)
496 return progs
499def _cygexpath(drive: Optional[str], path: str, expand_vars: bool = True) -> str:
500 if osp.isabs(path) and not drive:
501 # Invoked from `cygpath()` directly with `D:Apps\123`?
502 # It's an error, leave it alone just slashes)
503 p = path # convert to str if AnyPath given
504 else:
505 p = path and osp.normpath(osp.expandvars(osp.expanduser(path)) if expand_vars else path)
506 if osp.isabs(p):
507 if drive:
508 # Confusing, maybe a remote system should expand vars.
509 p = path
510 else:
511 p = cygpath(p)
512 elif drive:
513 p = "/proc/cygdrive/%s/%s" % (drive.lower(), p)
514 p_str = os.fspath(p) # ensure it is a str and not AnyPath
515 return p_str.replace("\\", "/")
518_cygpath_parsers: Tuple[Tuple[Pattern[str], Callable[..., str], bool], ...] = (
519 # See: https://msdn.microsoft.com/en-us/library/windows/desktop/aa365247(v=vs.85).aspx
520 # and: https://www.cygwin.com/cygwin-ug-net/using.html#unc-paths
521 (
522 re.compile(r"\\\\\?\\UNC\\([^\\]+)\\([^\\]+)(?:\\(.*))?"),
523 (lambda server, share, rest_path: "//%s/%s/%s" % (server, share, rest_path.replace("\\", "/"))),
524 False,
525 ),
526 (re.compile(r"\\\\\?\\(\w):[/\\](.*)"), (_cygexpath), False),
527 (re.compile(r"(\w):[/\\](.*)"), (_cygexpath), False),
528 (re.compile(r"file:(.*)", re.I), (lambda rest_path: rest_path), True),
529 (re.compile(r"(\w{2,}:.*)"), (lambda url: url), False), # remote URL, do nothing
530)
533def cygpath(path: str, expand_vars: bool = True) -> str:
534 """Use :meth:`git.cmd.Git.polish_url` instead, that works on any environment."""
535 path = os.fspath(path) # Ensure is str and not AnyPath.
536 # Fix to use Paths when 3.5 dropped. Or to be just str if only for URLs?
537 if not path.startswith(("/cygdrive", "//", "/proc/cygdrive")):
538 for regex, parser, recurse in _cygpath_parsers:
539 match = regex.match(path)
540 if match:
541 if parser is _cygexpath:
542 path = parser(*match.groups(), expand_vars=expand_vars)
543 else:
544 path = parser(*match.groups())
545 if recurse:
546 path = cygpath(path, expand_vars=expand_vars)
547 break
548 else:
549 path = _cygexpath(None, path, expand_vars=expand_vars)
551 return path
554_decygpath_regex = re.compile(r"(?:/proc)?/cygdrive/(\w)(/.*)?")
557def decygpath(path: PathLike) -> str:
558 path = os.fspath(path)
559 m = _decygpath_regex.match(path)
560 if m:
561 drive, rest_path = m.groups()
562 path = "%s:%s" % (drive.upper(), rest_path or "")
564 return path.replace("/", "\\")
567#: Store boolean flags denoting if a specific Git executable
568#: is from a Cygwin installation. TODO: use @functools.lru_cache(user_function)
569_is_cygwin_cache: Dict[str, Optional[bool]] = {}
572def _is_cygwin_git(git_executable: str) -> bool:
573 is_cygwin = _is_cygwin_cache.get(git_executable) # type: Optional[bool]
574 if is_cygwin is None:
575 is_cygwin = False
576 try:
577 git_dir = osp.dirname(git_executable)
578 if not git_dir:
579 res = py_where(git_executable)
580 git_dir = osp.dirname(res[0]) if res else ""
582 # Just a name given, not a real path.
583 uname_cmd = osp.join(git_dir, "uname")
585 if not (Path(uname_cmd).is_file() and os.access(uname_cmd, os.X_OK)):
586 _logger.debug(f"Failed checking if running in CYGWIN: {uname_cmd} is not an executable")
587 _is_cygwin_cache[git_executable] = is_cygwin
588 return is_cygwin
590 process = subprocess.Popen([uname_cmd], stdout=subprocess.PIPE, universal_newlines=True)
591 uname_out, _ = process.communicate()
592 # retcode = process.poll()
593 is_cygwin = "CYGWIN" in uname_out
594 except Exception as ex:
595 _logger.debug("Failed checking if running in CYGWIN due to: %r", ex)
596 _is_cygwin_cache[git_executable] = is_cygwin
598 return is_cygwin
601@overload
602def is_cygwin_git(git_executable: None) -> Literal[False]: ...
605@overload
606def is_cygwin_git(git_executable: PathLike) -> bool: ...
609def is_cygwin_git(git_executable: Union[None, PathLike]) -> bool:
610 _logger.debug(f"{sys.platform=}, {git_executable=}")
611 if sys.platform != "cygwin":
612 return False
613 elif git_executable is None:
614 return False
615 else:
616 return _is_cygwin_git(str(git_executable))
619def get_user_id() -> str:
620 """:return: String identifying the currently active system user as ``name@node``"""
621 return "%s@%s" % (getpass.getuser(), platform.node())
624def finalize_process(proc: Union["subprocess.Popen[Any]", "Git.AutoInterrupt"], **kwargs: Any) -> None:
625 """Wait for the process (clone, fetch, pull or push) and handle its errors
626 accordingly."""
627 # TODO: No close proc-streams??
628 proc.wait(**kwargs)
631@overload
632def expand_path(p: None, expand_vars: bool = ...) -> None: ...
635@overload
636def expand_path(p: PathLike, expand_vars: bool = ...) -> Optional[PathLike]:
637 # TODO: Support for Python 3.5 has been dropped, so these overloads can be improved.
638 ...
641def expand_path(p: Union[None, PathLike], expand_vars: bool = True) -> Optional[PathLike]:
642 if p is None:
643 return None
644 try:
645 if isinstance(p, Path):
646 return p.resolve()
647 expanded_path = osp.expanduser(os.fspath(p))
648 if expand_vars:
649 expanded_path = osp.expandvars(expanded_path)
650 return osp.normpath(osp.abspath(expanded_path))
651 except Exception:
652 return None
655def remove_password_if_present(cmdline: Sequence[str]) -> List[str]:
656 """Redact credentials in URLs and HTTP Authorization extra headers in a command line.
658 If nothing is found, this just returns the command line as-is.
660 This should be used for every log line that print a command line, as well as
661 exception messages.
662 """
663 new_cmdline = []
664 for index, to_parse in enumerate(cmdline):
665 new_cmdline.append(to_parse)
666 config_key, separator, header = to_parse.partition("=")
667 header_name, colon, _ = header.partition(":")
668 if (
669 separator
670 and colon
671 and config_key.lower().endswith(".extraheader")
672 and header_name.strip().lower() == "authorization"
673 ):
674 new_cmdline[index] = "%s%s%s%s *****" % (config_key, separator, header_name, colon)
675 continue
676 try:
677 url = urlsplit(to_parse)
678 # Remove password from the URL if present.
679 if url.password is None and url.username is None:
680 continue
682 # Match urllib.parse's userinfo boundary. Keeping the raw hostinfo
683 # preserves hostname case, IPv6 brackets, and port formatting.
684 _, _, hostinfo = url.netloc.rpartition("@")
685 redacted = "*****:*****" if url.password is not None else "*****"
686 url = url._replace(netloc=f"{redacted}@{hostinfo}")
687 new_cmdline[index] = urlunsplit(url)
688 except ValueError:
689 # This is not a valid URL.
690 continue
691 return new_cmdline
694# } END utilities
696# { Classes
699class RemoteProgress:
700 """Handler providing an interface to parse progress information emitted by
701 :manpage:`git-push(1)` and :manpage:`git-fetch(1)` and to dispatch callbacks
702 allowing subclasses to react to the progress."""
704 _num_op_codes: int = 9
705 (
706 BEGIN,
707 END,
708 COUNTING,
709 COMPRESSING,
710 WRITING,
711 RECEIVING,
712 RESOLVING,
713 FINDING_SOURCES,
714 CHECKING_OUT,
715 ) = [1 << x for x in range(_num_op_codes)]
716 STAGE_MASK = BEGIN | END
717 OP_MASK = ~STAGE_MASK
719 DONE_TOKEN = "done."
720 TOKEN_SEPARATOR = ", "
722 __slots__ = (
723 "_cur_line",
724 "_seen_ops",
725 "error_lines", # Lines that started with 'error:' or 'fatal:'.
726 "other_lines", # Lines not denoting progress (i.e.g. push-infos).
727 )
728 re_op_absolute = re.compile(r"(remote: )?([\w\s]+):\s+()(\d+)()(.*)")
729 re_op_relative = re.compile(r"(remote: )?([\w\s]+):\s+(\d+)% \((\d+)/(\d+)\)(.*)")
731 def __init__(self) -> None:
732 self._seen_ops: List[int] = []
733 self._cur_line: Optional[str] = None
734 self.error_lines: List[str] = []
735 self.other_lines: List[str] = []
737 def _parse_progress_line(self, line: AnyStr) -> Optional[object]:
738 """Parse progress information from the given line as retrieved by
739 :manpage:`git-push(1)` or :manpage:`git-fetch(1)`.
741 - Lines that do not contain progress info are stored in :attr:`other_lines`.
742 - Lines that seem to contain an error (i.e. start with ``error:`` or ``fatal:``)
743 are stored in :attr:`error_lines`.
745 The base implementation returns ``None``. Subclasses may return another
746 value for compatibility with existing overrides, but callers should treat
747 the return value as unspecified.
748 """
749 # handle
750 # Counting objects: 4, done.
751 # Compressing objects: 50% (1/2)
752 # Compressing objects: 100% (2/2)
753 # Compressing objects: 100% (2/2), done.
754 if isinstance(line, bytes): # mypy argues about ternary assignment.
755 line_str = line.decode("utf-8")
756 else:
757 line_str = line
758 self._cur_line = line_str
760 if self._cur_line.startswith(("error:", "fatal:")):
761 self.error_lines.append(self._cur_line)
762 return None
764 cur_count, max_count = None, None
765 match = self.re_op_relative.match(line_str)
766 if match is None:
767 match = self.re_op_absolute.match(line_str)
769 if not match:
770 self.line_dropped(line_str)
771 self.other_lines.append(line_str)
772 return None
773 # END could not get match
775 op_code = 0
776 _remote, op_name, _percent, cur_count, max_count, message = match.groups()
778 # Get operation ID.
779 if op_name == "Counting objects":
780 op_code |= self.COUNTING
781 elif op_name == "Compressing objects":
782 op_code |= self.COMPRESSING
783 elif op_name == "Writing objects":
784 op_code |= self.WRITING
785 elif op_name == "Receiving objects":
786 op_code |= self.RECEIVING
787 elif op_name == "Resolving deltas":
788 op_code |= self.RESOLVING
789 elif op_name == "Finding sources":
790 op_code |= self.FINDING_SOURCES
791 elif op_name == "Checking out files":
792 op_code |= self.CHECKING_OUT
793 else:
794 # Note: On Windows it can happen that partial lines are sent.
795 # Hence we get something like "CompreReceiving objects", which is
796 # a blend of "Compressing objects" and "Receiving objects".
797 # This can't really be prevented, so we drop the line verbosely
798 # to make sure we get informed in case the process spits out new
799 # commands at some point.
800 self.line_dropped(line_str)
801 # Note: Don't add this line to the other lines, as we have to silently
802 # drop it.
803 return None
804 # END handle op code
806 # Figure out stage.
807 if op_code not in self._seen_ops:
808 self._seen_ops.append(op_code)
809 op_code |= self.BEGIN
810 # END begin opcode
812 if message is None:
813 message = ""
814 # END message handling
816 message = message.strip()
817 if message.endswith(self.DONE_TOKEN):
818 op_code |= self.END
819 message = message[: -len(self.DONE_TOKEN)]
820 # END end message handling
821 message = message.strip(self.TOKEN_SEPARATOR)
823 self.update(
824 op_code,
825 cur_count and float(cur_count),
826 max_count and float(max_count),
827 message,
828 )
829 return None
831 def new_message_handler(self) -> Callable[[str], None]:
832 """
833 :return:
834 A progress handler suitable for :func:`~git.cmd.handle_process_output`,
835 passing lines on to this progress handler in a suitable format.
836 """
838 def handler(line: AnyStr) -> None:
839 self._parse_progress_line(line.rstrip())
841 # END handler
843 return handler
845 def line_dropped(self, line: str) -> None:
846 """Called whenever a line could not be understood and was therefore dropped."""
847 pass
849 def update(
850 self,
851 op_code: int,
852 cur_count: Union[str, float],
853 max_count: Union[str, float, None] = None,
854 message: str = "",
855 ) -> None:
856 """Called whenever the progress changes.
858 :param op_code:
859 Integer allowing to be compared against Operation IDs and stage IDs.
861 Stage IDs are :const:`BEGIN` and :const:`END`. :const:`BEGIN` will only be
862 set once for each Operation ID as well as :const:`END`. It may be that
863 :const:`BEGIN` and :const:`END` are set at once in case only one progress
864 message was emitted due to the speed of the operation. Between
865 :const:`BEGIN` and :const:`END`, none of these flags will be set.
867 Operation IDs are all held within the :const:`OP_MASK`. Only one Operation
868 ID will be active per call.
870 :param cur_count:
871 Current absolute count of items.
873 :param max_count:
874 The maximum count of items we expect. It may be ``None`` in case there is no
875 maximum number of items or if it is (yet) unknown.
877 :param message:
878 In case of the :const:`WRITING` operation, it contains the amount of bytes
879 transferred. It may possibly be used for other purposes as well.
881 :note:
882 You may read the contents of the current line in
883 :attr:`self._cur_line <_cur_line>`.
884 """
885 pass
888class CallableRemoteProgress(RemoteProgress):
889 """A :class:`RemoteProgress` implementation forwarding updates to any callable.
891 :note:
892 Like direct instances of :class:`RemoteProgress`, instances of this
893 :class:`CallableRemoteProgress` class are not themselves directly callable.
894 Rather, instances of this class wrap a callable and forward to it. This should
895 therefore not be confused with :class:`git.types.CallableProgress`.
896 """
898 __slots__ = ("_callable",)
900 def __init__(self, fn: Callable[..., Any]) -> None:
901 self._callable = fn
902 super().__init__()
904 def update(self, *args: Any, **kwargs: Any) -> None:
905 self._callable(*args, **kwargs)
908class _DeprecatedActorNameEmailRegex:
909 _pattern = re.compile(r"(.*) <(.*?)>")
911 def __get__(self, _instance: Any, _owner: Any) -> Pattern[str]:
912 warnings.warn(
913 "Actor.name_email_regex is deprecated and will be removed in GitPython 4.0.0 because searching long "
914 "malformed strings with it can take quadratic time. Use Actor.from_string() to parse actor identities, "
915 "or Actor(name, email) when the fields are already separate.",
916 DeprecationWarning,
917 stacklevel=2,
918 )
919 return self._pattern
922class Actor:
923 """Actors hold information about a person acting on the repository. They can be
924 committers and authors or anything with a name and an email as mentioned in the git
925 log entries."""
927 name_email_regex = _DeprecatedActorNameEmailRegex()
929 # ENVIRONMENT VARIABLES
930 # These are read when creating new commits.
931 env_author_name = "GIT_AUTHOR_NAME"
932 env_author_email = "GIT_AUTHOR_EMAIL"
933 env_committer_name = "GIT_COMMITTER_NAME"
934 env_committer_email = "GIT_COMMITTER_EMAIL"
936 # CONFIGURATION KEYS
937 conf_name = "name"
938 conf_email = "email"
940 __slots__ = ("name", "email")
942 def __init__(self, name: Optional[str], email: Optional[str]) -> None:
943 self.name = name
944 self.email = email
946 def __eq__(self, other: Any) -> bool:
947 return self.name == other.name and self.email == other.email
949 def __ne__(self, other: Any) -> bool:
950 return not (self == other)
952 def __hash__(self) -> int:
953 return hash((self.name, self.email))
955 def __str__(self) -> str:
956 return self.name if self.name else ""
958 def __repr__(self) -> str:
959 return '<git.Actor "%s <%s>">' % (self.name, self.email)
961 @classmethod
962 def from_string(cls: Type[T_Actor], string: str) -> T_Actor:
963 """Create an :class:`Actor` from a string.
965 :param string:
966 The string, which is expected to be in regular git format::
968 John Doe <jdoe@example.com>
970 :return:
971 :class:`Actor`
972 """
973 line = string.partition("\n")[0]
974 left_bracket = line.find("<")
975 right_bracket = line.find(">", left_bracket + 1)
976 if left_bracket >= 0 and right_bracket >= 0:
977 return cls(line[:left_bracket].rstrip(), line[left_bracket + 1 : right_bracket])
979 # Assume the best and use the whole string as name.
980 return cls(string, None)
982 _from_string = from_string
984 @classmethod
985 def _main_actor(
986 cls,
987 env_name: str,
988 env_email: str,
989 config_reader: Union[None, "GitConfigParser", "SectionConstraint[GitConfigParser]"] = None,
990 ) -> "Actor":
991 actor = Actor("", "")
992 user_id = None # We use this to avoid multiple calls to getpass.getuser().
994 def default_email() -> str:
995 nonlocal user_id
996 if not user_id:
997 user_id = get_user_id()
998 return user_id
1000 def default_name() -> str:
1001 return default_email().split("@")[0]
1003 for attr, evar, cvar, default in (
1004 ("name", env_name, cls.conf_name, default_name),
1005 ("email", env_email, cls.conf_email, default_email),
1006 ):
1007 try:
1008 val = os.environ[evar]
1009 setattr(actor, attr, val)
1010 except KeyError:
1011 if config_reader is not None:
1012 try:
1013 val = config_reader.get("user", cvar)
1014 except Exception:
1015 val = default()
1016 setattr(actor, attr, val)
1017 # END config-reader handling
1018 if not getattr(actor, attr):
1019 setattr(actor, attr, default())
1020 # END handle name
1021 # END for each item to retrieve
1022 return actor
1024 @classmethod
1025 def committer(
1026 cls, config_reader: Union[None, "GitConfigParser", "SectionConstraint[GitConfigParser]"] = None
1027 ) -> "Actor":
1028 """
1029 :return:
1030 :class:`Actor` instance corresponding to the configured committer. It
1031 behaves similar to the git implementation, such that the environment will
1032 override configuration values of `config_reader`. If no value is set at all,
1033 it will be generated.
1035 :param config_reader:
1036 ConfigReader to use to retrieve the values from in case they are not set in
1037 the environment.
1038 """
1039 return cls._main_actor(cls.env_committer_name, cls.env_committer_email, config_reader)
1041 @classmethod
1042 def author(
1043 cls, config_reader: Union[None, "GitConfigParser", "SectionConstraint[GitConfigParser]"] = None
1044 ) -> "Actor":
1045 """Same as :meth:`committer`, but defines the main author. It may be specified
1046 in the environment, but defaults to the committer."""
1047 return cls._main_actor(cls.env_author_name, cls.env_author_email, config_reader)
1050class Stats:
1051 """Represents stat information as presented by git at the end of a merge. It is
1052 created from the output of a diff operation.
1054 Example::
1056 c = Commit( sha1 )
1057 s = c.stats
1058 s.total # full-stat-dict
1059 s.files # dict( filepath : stat-dict )
1061 ``stat-dict``
1063 A dictionary with the following keys and values::
1065 deletions = number of deleted lines as int
1066 insertions = number of inserted lines as int
1067 lines = total number of lines changed as int, or deletions + insertions
1068 change_type = type of change as str, A|C|D|M|R|T|U|X|B
1070 ``full-stat-dict``
1072 In addition to the items in the stat-dict, it features additional information::
1074 files = number of changed files as int
1075 """
1077 __slots__ = ("total", "files")
1079 def __init__(self, total: Total_TD, files: Dict[PathLike, Files_TD]) -> None:
1080 self.total = total
1081 self.files = files
1083 @classmethod
1084 def _list_from_string(cls, repo: "Repo", text: str) -> "Stats":
1085 """Create a :class:`Stats` object from output retrieved by
1086 :manpage:`git-diff(1)`.
1088 :return:
1089 :class:`git.Stats`
1090 """
1092 hsh: HSH_TD = {
1093 "total": {"insertions": 0, "deletions": 0, "lines": 0, "files": 0},
1094 "files": {},
1095 }
1096 for line in text.splitlines():
1097 (change_type, raw_insertions, raw_deletions, filename) = line.split("\t")
1098 insertions = raw_insertions != "-" and int(raw_insertions) or 0
1099 deletions = raw_deletions != "-" and int(raw_deletions) or 0
1100 hsh["total"]["insertions"] += insertions
1101 hsh["total"]["deletions"] += deletions
1102 hsh["total"]["lines"] += insertions + deletions
1103 hsh["total"]["files"] += 1
1104 files_dict: Files_TD = {
1105 "insertions": insertions,
1106 "deletions": deletions,
1107 "lines": insertions + deletions,
1108 "change_type": change_type,
1109 }
1110 hsh["files"][filename] = files_dict
1111 return Stats(hsh["total"], hsh["files"])
1114class IndexFileSHA1Writer:
1115 """Wrapper around a file-like object that remembers the SHA1 of the data written to
1116 it. It will write a sha when the stream is closed or if asked for explicitly using
1117 :meth:`write_sha`.
1119 Only useful to the index file.
1121 :note:
1122 Based on the dulwich project.
1123 """
1125 __slots__ = ("f", "sha1")
1127 def __init__(self, f: IO[bytes]) -> None:
1128 self.f = f
1129 self.sha1 = make_sha(b"")
1131 def write(self, data: bytes) -> int:
1132 self.sha1.update(data)
1133 return self.f.write(data)
1135 def write_sha(self) -> bytes:
1136 sha = self.sha1.digest()
1137 self.f.write(sha)
1138 return sha
1140 def close(self) -> bytes:
1141 sha = self.write_sha()
1142 self.f.close()
1143 return sha
1145 def tell(self) -> int:
1146 return self.f.tell()
1149class LockFile:
1150 """Provides methods to obtain, check for, and release a file based lock which
1151 should be used to handle concurrent access to the same file.
1153 As we are a utility class to be derived from, we only use protected methods.
1155 Locks will automatically be released on destruction.
1156 """
1158 __slots__ = ("_file_path", "_owns_lock")
1160 def __init__(self, file_path: PathLike) -> None:
1161 self._file_path = file_path
1162 self._owns_lock = False
1164 def __del__(self) -> None:
1165 self._release_lock()
1167 def _lock_file_path(self) -> str:
1168 """:return: Path to lockfile"""
1169 return "%s.lock" % (self._file_path)
1171 def _has_lock(self) -> bool:
1172 """
1173 :return:
1174 True if we have a lock and if the lockfile still exists
1176 :raise AssertionError:
1177 If our lock-file does not exist.
1178 """
1179 return self._owns_lock
1181 def _obtain_lock_or_raise(self) -> None:
1182 """Create a lock file as flag for other instances, mark our instance as
1183 lock-holder.
1185 :raise IOError:
1186 If a lock was already present or a lock file could not be written.
1187 """
1188 if self._has_lock():
1189 return
1190 lock_file = self._lock_file_path()
1191 # Create the lock in one step. Checking for it first would allow another
1192 # holder to create it between the check and the open.
1193 try:
1194 if sys.platform == "win32":
1195 if "\0" in lock_file:
1196 raise ValueError("embedded null character")
1198 import ctypes
1199 from ctypes import wintypes
1201 # Unlike POSIX, Windows follows dangling symlinks even with O_EXCL.
1202 # Open the reparse point itself so an existing link is rejected.
1203 # Call the Unicode API directly: older _winapi.CreateFile wrappers
1204 # use the ANSI API and can create a lock under the wrong filename.
1205 kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
1206 create_file = kernel32.CreateFileW
1207 create_file.argtypes = (
1208 wintypes.LPCWSTR,
1209 wintypes.DWORD,
1210 wintypes.DWORD,
1211 wintypes.LPVOID,
1212 wintypes.DWORD,
1213 wintypes.DWORD,
1214 wintypes.HANDLE,
1215 )
1216 create_file.restype = wintypes.HANDLE
1217 close_handle = kernel32.CloseHandle
1218 close_handle.argtypes = (wintypes.HANDLE,)
1219 close_handle.restype = wintypes.BOOL
1220 handle = create_file(
1221 lock_file,
1222 0x40000000, # GENERIC_WRITE
1223 0,
1224 None,
1225 1, # CREATE_NEW
1226 0x00200000, # FILE_FLAG_OPEN_REPARSE_POINT
1227 None,
1228 )
1229 if handle == wintypes.HANDLE(-1).value:
1230 raise ctypes.WinError(ctypes.get_last_error())
1231 if not close_handle(handle):
1232 raise ctypes.WinError(ctypes.get_last_error())
1233 else:
1234 fd = os.open(lock_file, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
1235 os.close(fd)
1236 except FileExistsError as e:
1237 raise OSError(
1238 "Lock for file %r did already exist, delete %r in case the lock is illegal"
1239 % (self._file_path, lock_file)
1240 ) from e
1241 except OSError as e:
1242 raise OSError(str(e)) from e
1244 self._owns_lock = True
1246 def _obtain_lock(self) -> None:
1247 """The default implementation will raise if a lock cannot be obtained.
1249 Subclasses may override this method to provide a different implementation.
1250 """
1251 return self._obtain_lock_or_raise()
1253 def _release_lock(self) -> None:
1254 """Release our lock if we have one."""
1255 if not self._has_lock():
1256 return
1258 # If someone removed our file beforehand, lets just flag this issue instead of
1259 # failing, to make it more usable.
1260 lfp = self._lock_file_path()
1261 try:
1262 rmfile(lfp)
1263 except OSError:
1264 pass
1265 self._owns_lock = False
1268class BlockingLockFile(LockFile):
1269 """The lock file will block until a lock could be obtained, or fail after a
1270 specified timeout.
1272 :note:
1273 Only an existing lock causes retries. Other errors, including a removed or
1274 unwritable parent directory, are raised immediately.
1275 """
1277 __slots__ = ("_check_interval", "_max_block_time")
1279 def __init__(
1280 self,
1281 file_path: PathLike,
1282 check_interval_s: float = 0.3,
1283 max_block_time_s: int = sys.maxsize,
1284 ) -> None:
1285 """Configure the instance.
1287 :param check_interval_s:
1288 Period of time to sleep until the lock is checked the next time.
1289 By default, it waits a nearly unlimited time.
1291 :param max_block_time_s:
1292 Maximum amount of seconds we may lock.
1293 """
1294 super().__init__(file_path)
1295 self._check_interval = check_interval_s
1296 self._max_block_time = max_block_time_s
1298 def _obtain_lock(self) -> None:
1299 """This method blocks until it obtained the lock, or raises :exc:`IOError` if it
1300 ran out of time or lock creation failed for a reason other than an existing lock.
1302 If this method returns, you are guaranteed to own the lock.
1303 """
1304 starttime = time.time()
1305 maxtime = starttime + float(self._max_block_time)
1306 while True:
1307 try:
1308 super()._obtain_lock()
1309 except OSError as e:
1310 # LockFile retains the creation error as the cause of its OSError.
1311 # Only an existing lock is contention; other errors must not be retried.
1312 if not isinstance(e.__cause__, FileExistsError):
1313 raise
1315 # Abort if the directory containing the existing lock has disappeared.
1316 curtime = time.time()
1317 if not osp.isdir(osp.dirname(self._lock_file_path()) or osp.curdir):
1318 msg = "Directory containing the lockfile %r was not readable anymore after waiting %g seconds" % (
1319 self._lock_file_path(),
1320 curtime - starttime,
1321 )
1322 raise OSError(msg) from e
1323 # END handle missing directory
1325 if curtime >= maxtime:
1326 msg = "Waited %g seconds for lock at %r" % (
1327 maxtime - starttime,
1328 self._lock_file_path(),
1329 )
1330 raise OSError(msg) from e
1331 # END abort if we wait too long
1332 time.sleep(self._check_interval)
1333 else:
1334 break
1335 # END endless loop
1338class IterableList(List[T_IterableObj]): # type: ignore[type-var]
1339 """List of iterable objects allowing to query an object by id or by named index::
1341 heads = repo.heads
1342 heads.master
1343 heads['master']
1344 heads[0]
1346 Iterable parent objects:
1348 * :class:`Commit <git.objects.Commit>`
1349 * :class:`Submodule <git.objects.submodule.base.Submodule>`
1350 * :class:`Reference <git.refs.reference.Reference>`
1351 * :class:`FetchInfo <git.remote.FetchInfo>`
1352 * :class:`PushInfo <git.remote.PushInfo>`
1354 Iterable via inheritance:
1356 * :class:`Head <git.refs.head.Head>`
1357 * :class:`TagReference <git.refs.tag.TagReference>`
1358 * :class:`RemoteReference <git.refs.remote.RemoteReference>`
1360 This requires an ``id_attribute`` name to be set which will be queried from its
1361 contained items to have a means for comparison.
1363 A prefix can be specified which is to be used in case the id returned by the items
1364 always contains a prefix that does not matter to the user, so it can be left out.
1365 """
1367 __slots__ = ("_id_attr", "_prefix")
1369 def __new__(cls, id_attr: str, prefix: str = "") -> "IterableList[T_IterableObj]":
1370 return super().__new__(cls)
1372 def __init__(self, id_attr: str, prefix: str = "") -> None:
1373 super().__init__()
1374 self._id_attr = id_attr
1375 self._prefix = prefix
1377 def __contains__(self, attr: object) -> bool:
1378 # First try identity match for performance.
1379 try:
1380 rval = list.__contains__(self, attr)
1381 if rval:
1382 return rval
1383 except (AttributeError, TypeError):
1384 pass
1385 # END handle match
1387 # Otherwise make a full name search.
1388 try:
1389 self.__getattr__(cast(str, attr)) # Use cast to silence mypy.
1390 return True
1391 except (AttributeError, TypeError):
1392 return False
1393 # END handle membership
1395 def __getattr__(self, attr: str) -> T_IterableObj:
1396 attr = self._prefix + attr
1397 for item in self:
1398 if getattr(item, self._id_attr) == attr:
1399 return item
1400 # END for each item
1401 raise AttributeError(f"No item found with id {attr}")
1403 def __getitem__( # type: ignore[override] # pyright: ignore[reportIncompatibleMethodOverride]
1404 self, index: Union[SupportsIndex, int, slice, str]
1405 ) -> T_IterableObj:
1406 if isinstance(index, int):
1407 return list.__getitem__(self, index)
1408 elif isinstance(index, slice):
1409 raise ValueError("Index should be an int or str")
1410 else:
1411 try:
1412 return self.__getattr__(cast(str, index))
1413 except AttributeError as e:
1414 raise IndexError(f"No item found with id {self._prefix}{index}") from e
1415 # END handle getattr
1417 def __delitem__(self, index: Union[SupportsIndex, int, slice, str]) -> None:
1418 delindex = cast(int, index)
1419 if isinstance(index, str):
1420 delindex = -1
1421 name = self._prefix + index
1422 for i, item in enumerate(self):
1423 if getattr(item, self._id_attr) == name:
1424 delindex = i
1425 break
1426 # END search index
1427 # END for each item
1428 if delindex == -1:
1429 raise IndexError("Item with name %s not found" % name)
1430 # END handle error
1431 # END get index to delete
1432 list.__delitem__(self, delindex)
1435@runtime_checkable
1436class IterableObj(Protocol):
1437 """Defines an interface for iterable items, so there is a uniform way to retrieve
1438 and iterate items within the git repository.
1440 Subclasses:
1442 * :class:`Submodule <git.objects.submodule.base.Submodule>`
1443 * :class:`Commit <git.objects.Commit>`
1444 * :class:`Reference <git.refs.reference.Reference>`
1445 * :class:`PushInfo <git.remote.PushInfo>`
1446 * :class:`FetchInfo <git.remote.FetchInfo>`
1447 * :class:`Remote <git.remote.Remote>`
1448 """
1450 __slots__ = ()
1452 _id_attribute_: str
1454 @classmethod
1455 @abstractmethod
1456 def iter_items(cls, repo: "Repo", *args: Any, **kwargs: Any) -> Iterator[T_IterableObj]:
1457 # Return-typed to be compatible with subtypes e.g. Remote.
1458 """Find (all) items of this type.
1460 Subclasses can specify `args` and `kwargs` differently, and may use them for
1461 filtering. However, when the method is called with no additional positional or
1462 keyword arguments, subclasses are obliged to to yield all items.
1464 :return:
1465 Iterator yielding Items
1466 """
1467 raise NotImplementedError("To be implemented by Subclass")
1469 @classmethod
1470 def list_items(cls, repo: "Repo", *args: Any, **kwargs: Any) -> IterableList[T_IterableObj]:
1471 """Find (all) items of this type and collect them into a list.
1473 For more information about the arguments, see :meth:`iter_items`.
1475 :note:
1476 Favor the :meth:`iter_items` method as it will avoid eagerly collecting all
1477 items. When there are many items, that can slow performance and increase
1478 memory usage.
1480 :return:
1481 list(Item,...) list of item instances
1482 """
1483 out_list: IterableList[T_IterableObj] = IterableList(cls._id_attribute_)
1484 out_list.extend(cls.iter_items(repo, *args, **kwargs))
1485 return out_list
1488class IterableClassWatcher(type):
1489 """Metaclass that issues :exc:`DeprecationWarning` when :class:`git.util.Iterable`
1490 is subclassed."""
1492 def __init__(cls, name: str, bases: Tuple[type, ...], clsdict: Dict[str, Any]) -> None:
1493 super().__init__(name, bases, clsdict)
1494 for base in bases:
1495 if type(base) is IterableClassWatcher:
1496 warnings.warn(
1497 f"GitPython Iterable subclassed by {name}."
1498 " Iterable is deprecated due to naming clash since v3.1.18"
1499 " and will be removed in 4.0.0."
1500 " Use IterableObj instead.",
1501 DeprecationWarning,
1502 stacklevel=2,
1503 )
1506class Iterable(metaclass=IterableClassWatcher):
1507 """Deprecated, use :class:`IterableObj` instead.
1509 Defines an interface for iterable items, so there is a uniform way to retrieve
1510 and iterate items within the git repository.
1511 """
1513 __slots__ = ()
1515 _id_attribute_ = "attribute that most suitably identifies your instance"
1517 @classmethod
1518 def iter_items(cls, repo: "Repo", *args: Any, **kwargs: Any) -> Any:
1519 """Deprecated, use :class:`IterableObj` instead.
1521 Find (all) items of this type.
1523 See :meth:`IterableObj.iter_items` for details on usage.
1525 :return:
1526 Iterator yielding Items
1527 """
1528 raise NotImplementedError("To be implemented by Subclass")
1530 @classmethod
1531 def list_items(cls, repo: "Repo", *args: Any, **kwargs: Any) -> Any:
1532 """Deprecated, use :class:`IterableObj` instead.
1534 Find (all) items of this type and collect them into a list.
1536 See :meth:`IterableObj.list_items` for details on usage.
1538 :return:
1539 list(Item,...) list of item instances
1540 """
1541 out_list: Any = IterableList(cls._id_attribute_)
1542 out_list.extend(cls.iter_items(repo, *args, **kwargs))
1543 return out_list
1546# } END classes