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