Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/git/cmd.py: 57%
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/
6from __future__ import annotations
8__all__ = ["GitMeta", "Git"]
10import contextlib
11import io
12import itertools
13import logging
14import os
15import re
16import signal
17import subprocess
18import sys
19import threading
20import time
21import warnings
22from subprocess import DEVNULL, PIPE, Popen
23from textwrap import dedent
25# typing ---------------------------------------------------------------------------
26from typing import (
27 IO,
28 TYPE_CHECKING,
29 Any,
30 AnyStr,
31 BinaryIO,
32 Callable,
33 Dict,
34 Iterator,
35 List,
36 Mapping,
37 Optional,
38 Sequence,
39 TextIO,
40 Tuple,
41 Union,
42 cast,
43 overload,
44)
46from git.compat import defenc, force_bytes, safe_decode
47from git.exc import (
48 CommandError,
49 GitCommandError,
50 GitCommandNotFound,
51 UnsafeOptionError,
52 UnsafeProtocolError,
53)
54from git.util import (
55 cygpath,
56 expand_path,
57 is_cygwin_git,
58 patch_env,
59 remove_password_if_present,
60 stream_copy,
61)
63if sys.version_info >= (3, 10):
64 from typing import TypeAlias
65else:
66 from typing_extensions import TypeAlias
68from git.types import TBD, Literal, PathLike
70if TYPE_CHECKING:
71 from git.diff import DiffIndex
72 from git.repo.base import Repo
74# ---------------------------------------------------------------------------------
76execute_kwargs = {
77 "istream",
78 "with_extended_output",
79 "with_exceptions",
80 "as_process",
81 "output_stream",
82 "stdout_as_string",
83 "kill_after_timeout",
84 "with_stdout",
85 "universal_newlines",
86 "shell",
87 "env",
88 "max_chunk_size",
89 "strip_newline_in_stdout",
90}
92_logger = logging.getLogger(__name__)
95# ==============================================================================
96## @name Utilities
97# ------------------------------------------------------------------------------
98# Documentation
99## @{
102def _kill_process(pid: int) -> bool:
103 """Kill a POSIX process and its descendants, returning whether it was killed."""
104 # Collect descendants before signalling, while their parent PIDs still identify them.
105 pids = [pid]
106 for parent_pid in pids:
107 try:
108 try:
109 p = Popen(["pgrep", "-P", str(parent_pid)], stdout=PIPE)
110 except FileNotFoundError:
111 # POSIX ps does not support selecting by parent PID.
112 ps_args = ["ps", "-ef"] if sys.platform == "cygwin" else ["ps", "-A", "-o", "pid=", "-o", "ppid="]
113 with Popen(ps_args, stdout=PIPE) as p:
114 if p.stdout is not None:
115 for line in p.stdout:
116 fields = line.split()
117 if sys.platform == "cygwin":
118 fields = fields[1:3] # ps -ef starts with UID, PID, PPID.
119 if len(fields) == 2 and all(field.isdigit() for field in fields):
120 if int(fields[1]) == parent_pid:
121 pids.append(int(fields[0]))
122 else:
123 with p:
124 if p.stdout is not None:
125 for line in p.stdout:
126 if line.strip().isdigit():
127 pids.append(int(line))
128 except OSError as ex:
129 _logger.info("Unable to enumerate child processes: %r", ex)
130 killed = False
131 if sys.platform != "win32":
132 for process_pid in pids:
133 try:
134 os.kill(process_pid, signal.SIGKILL)
135 killed = killed or process_pid == pid
136 except OSError:
137 pass
138 return killed
141def handle_process_output(
142 process: Union["Git.AutoInterrupt", Popen],
143 stdout_handler: Union[
144 None,
145 Callable[[AnyStr], None],
146 Callable[[List[AnyStr]], None],
147 Callable[[bytes, "Repo", "DiffIndex"], None],
148 ],
149 stderr_handler: Union[None, Callable[[AnyStr], None], Callable[[List[AnyStr]], None]],
150 finalizer: Union[None, Callable[[Union[Popen, "Git.AutoInterrupt"]], None]] = None,
151 decode_streams: bool = True,
152 kill_after_timeout: Union[None, float] = None,
153) -> None:
154 R"""Register for notifications to learn that process output is ready to read, and
155 dispatch lines to the respective line handlers.
157 This function returns once the finalizer returns.
159 :param process:
160 :class:`subprocess.Popen` instance.
162 :param stdout_handler:
163 f(stdout_line_string), or ``None``.
165 :param stderr_handler:
166 f(stderr_line_string), or ``None``.
168 :param finalizer:
169 f(proc) - wait for proc to finish.
171 :param decode_streams:
172 Assume stdout/stderr streams are binary and decode them before pushing their
173 contents to handlers.
175 This defaults to ``True``. Set it to ``False`` if:
177 - ``universal_newlines == True``, as then streams are in text mode, or
178 - decoding must happen later, such as for :class:`~git.diff.Diff`\s.
180 :param kill_after_timeout:
181 :class:`float` or ``None``, Default = ``None``
183 To specify a timeout in seconds for the git command, after which the process
184 should be killed.
185 """
187 # Use 2 "pump" threads and wait for both to finish.
188 def pump_stream(
189 cmdline: List[str],
190 name: str,
191 stream: Union[BinaryIO, TextIO],
192 is_decode: bool,
193 handler: Union[None, Callable[[Union[bytes, str]], None]],
194 ) -> None:
195 try:
196 for line in stream:
197 if handler:
198 if is_decode:
199 assert isinstance(line, bytes)
200 line_str = line.decode(defenc)
201 handler(line_str)
202 else:
203 handler(line)
205 except Exception as ex:
206 _logger.error(f"Pumping {name!r} of cmd({remove_password_if_present(cmdline)}) failed due to: {ex!r}")
207 if not (isinstance(ex, ValueError) and stream.closed):
208 # Only reraise if the error was not due to the stream closing.
209 raise CommandError([f"<{name}-pump>"] + remove_password_if_present(cmdline), ex) from ex
210 finally:
211 stream.close()
213 if hasattr(process, "proc"):
214 process = cast("Git.AutoInterrupt", process)
215 cmdline: str | Tuple[str, ...] | List[str] = getattr(process.proc, "args", "")
216 p_stdout = process.proc.stdout if process.proc else None
217 p_stderr = process.proc.stderr if process.proc else None
218 else:
219 process = cast(Popen, process) # type: ignore[redundant-cast]
220 cmdline = getattr(process, "args", "")
221 p_stdout = process.stdout
222 p_stderr = process.stderr
224 if not isinstance(cmdline, (tuple, list)):
225 cmdline = cmdline.split()
227 pumps: List[Tuple[str, IO, Callable[..., None] | None]] = []
228 if p_stdout:
229 pumps.append(("stdout", p_stdout, stdout_handler))
230 if p_stderr:
231 pumps.append(("stderr", p_stderr, stderr_handler))
233 threads: List[threading.Thread] = []
235 for name, stream, handler in pumps:
236 t = threading.Thread(target=pump_stream, args=(cmdline, name, stream, decode_streams, handler))
237 t.daemon = True
238 t.start()
239 threads.append(t)
241 # Wait for output handlers to finish before finalizing.
242 # If the child needs stdin, the caller must arrange to feed/close it
243 # before this call or concurrently; this function only drains output.
244 deadline = None if kill_after_timeout is None else time.monotonic() + kill_after_timeout
245 for t in threads:
246 t.join(timeout=None if deadline is None else max(0, deadline - time.monotonic()))
247 if t.is_alive():
248 if isinstance(process, Git.AutoInterrupt):
249 if sys.platform != "win32" and process.proc is not None and process.proc.poll() is None:
250 _kill_process(process.proc.pid)
251 process._terminate()
252 else: # Don't want to deal with the other case.
253 raise RuntimeError(
254 "Thread join() timed out in cmd.handle_process_output()."
255 f" kill_after_timeout={kill_after_timeout} seconds"
256 )
257 process._timeout_error = (
258 f"error: process killed because it timed out. kill_after_timeout={kill_after_timeout} seconds"
259 )
260 break
262 if finalizer:
263 finalizer(process)
266safer_popen: Callable[..., Popen]
268if sys.platform == "win32":
270 def _safer_popen_windows(
271 command: Union[str, Sequence[Any]],
272 *,
273 shell: bool = False,
274 env: Optional[Mapping[str, str]] = None,
275 **kwargs: Any,
276 ) -> Popen:
277 """Call :class:`subprocess.Popen` on Windows but don't include a CWD in the
278 search.
280 This avoids an untrusted search path condition where a file like ``git.exe`` in
281 a malicious repository would be run when GitPython operates on the repository.
282 The process using GitPython may have an untrusted repository's working tree as
283 its current working directory. Some operations may temporarily change to that
284 directory before running a subprocess. In addition, while by default GitPython
285 does not run external commands with a shell, it can be made to do so, in which
286 case the CWD of the subprocess, which GitPython usually sets to a repository
287 working tree, can itself be searched automatically by the shell. This wrapper
288 covers all those cases.
290 :note:
291 This currently works by setting the
292 :envvar:`NoDefaultCurrentDirectoryInExePath` environment variable during
293 subprocess creation. It also takes care of passing Windows-specific process
294 creation flags, but that is unrelated to path search.
296 :note:
297 The current implementation contains a race condition on :attr:`os.environ`.
298 GitPython isn't thread-safe, but a program using it on one thread should
299 ideally be able to mutate :attr:`os.environ` on another, without
300 unpredictable results. See comments in:
301 https://github.com/gitpython-developers/GitPython/pull/1650
302 """
303 # CREATE_NEW_PROCESS_GROUP is needed for some ways of killing it afterwards.
304 # https://docs.python.org/3/library/subprocess.html#subprocess.Popen.send_signal
305 # https://docs.python.org/3/library/subprocess.html#subprocess.CREATE_NEW_PROCESS_GROUP
306 creationflags = subprocess.CREATE_NO_WINDOW | subprocess.CREATE_NEW_PROCESS_GROUP
308 # When using a shell, the shell is the direct subprocess, so the variable must
309 # be set in its environment, to affect its search behavior.
310 if shell:
311 # The original may be immutable, or the caller may reuse it. Mutate a copy.
312 env = {} if env is None else dict(env)
313 env["NoDefaultCurrentDirectoryInExePath"] = "1" # The "1" can be any value.
315 # When not using a shell, the current process does the search in a
316 # CreateProcessW API call, so the variable must be set in our environment. With
317 # a shell, that's unnecessary if https://github.com/python/cpython/issues/101283
318 # is patched. In Python versions where it is unpatched, in the rare case the
319 # ComSpec environment variable is unset, the search for the shell itself is
320 # unsafe. Setting NoDefaultCurrentDirectoryInExePath in all cases, as done here,
321 # is simpler and protects against that. (As above, the "1" can be any value.)
322 with patch_env("NoDefaultCurrentDirectoryInExePath", "1"):
323 return Popen(
324 command,
325 shell=shell,
326 env=env,
327 creationflags=creationflags,
328 **kwargs,
329 )
331 safer_popen = _safer_popen_windows
332else:
333 safer_popen = Popen
336def dashify(string: str) -> str:
337 return string.replace("_", "-")
340def slots_to_dict(self: "Git", exclude: Sequence[str] = ()) -> Dict[str, Any]:
341 return {s: getattr(self, s) for s in self.__slots__ if s not in exclude}
344def dict_to_slots_and__excluded_are_none(self: object, d: Mapping[str, Any], excluded: Sequence[str] = ()) -> None:
345 for k, v in d.items():
346 setattr(self, k, v)
347 for k in excluded:
348 setattr(self, k, None)
351## -- End Utilities -- @}
354class _AutoInterrupt:
355 """Process wrapper that terminates the wrapped process on finalization.
357 This kills/interrupts the stored process instance once this instance goes out of
358 scope. It is used to prevent processes piling up in case iterators stop reading.
360 All attributes are wired through to the contained process object.
362 The wait method is overridden to perform automatic status code checking and possibly
363 raise.
364 """
366 __slots__ = ("proc", "args", "status", "_timeout_error")
368 # If this is non-zero it will override any status code during _terminate, used
369 # to prevent race conditions in testing.
370 _status_code_if_terminate: int = 0
372 def __init__(self, proc: Union[None, subprocess.Popen], args: Any) -> None:
373 self.proc = proc
374 self.args = args
375 self.status: Union[int, None] = None
376 self._timeout_error: Optional[str] = None
378 def _terminate(self) -> None:
379 """Terminate the underlying process."""
380 if self.proc is None:
381 return
383 proc = self.proc
384 self.proc = None
385 try:
386 if proc.stdin:
387 # A timed-out process may already have exited before input is flushed.
388 with contextlib.suppress(BrokenPipeError):
389 proc.stdin.close()
390 # Did the process finish already so we have a return code?
391 try:
392 if proc.poll() is not None:
393 self.status = self._status_code_if_terminate or proc.poll()
394 return
395 except OSError as ex:
396 _logger.info("Ignored error after process had died: %r", ex)
398 # It can be that nothing really exists anymore...
399 if getattr(os, "kill", None) is None:
400 return
402 # Try to kill it.
403 try:
404 proc.terminate()
405 except (OSError, AttributeError) as ex:
406 # On interpreter shutdown (notably on Windows), parts of the stdlib used by
407 # subprocess can already be torn down (e.g. `subprocess._winapi` becomes None),
408 # which can cause AttributeError during terminate(). In that case, we prefer
409 # to silently ignore to avoid noisy "Exception ignored in: __del__" messages.
410 _logger.info("Ignored error while terminating process: %r", ex)
411 return
412 # END exception handling
413 finally:
414 if proc.stdout:
415 proc.stdout.close()
416 if proc.stderr:
417 proc.stderr.close()
419 try:
420 status = proc.wait()
421 self.status = self._status_code_if_terminate or status
422 except (OSError, AttributeError) as ex:
423 _logger.info("Ignored error while waiting for terminated process: %r", ex)
425 def __del__(self) -> None:
426 self._terminate()
428 def __getattr__(self, attr: str) -> Any:
429 return getattr(self.proc, attr)
431 # TODO: Bad choice to mimic `proc.wait()` but with different args.
432 def wait(self, stderr: Union[None, str, bytes] = b"") -> int:
433 """Wait for the process and return its status code.
435 :param stderr:
436 Previously read value of stderr, in case stderr is already closed.
438 :warn:
439 May deadlock if output or error pipes are used and not handled separately.
441 :raise git.exc.GitCommandError:
442 If the return status is not 0 or output handling timed out.
443 """
444 status: Union[int, None]
445 if self.proc is not None:
446 status = self.proc.wait()
447 p_stderr = self.proc.stderr
448 else: # Assume the underlying proc was killed earlier or never existed.
449 status = self.status
450 p_stderr = None
452 stderr_b = force_bytes(data=stderr, encoding="utf-8") or b""
453 if self._timeout_error is not None:
454 stderr_b += (b"\n" if stderr_b else b"") + self._timeout_error.encode("utf-8")
455 status = status or 1
457 def read_all_from_possibly_closed_stream(stream: Union[IO[bytes], None]) -> bytes:
458 if stream:
459 try:
460 return stderr_b + force_bytes(stream.read())
461 except (OSError, ValueError):
462 return stderr_b or b""
463 else:
464 return stderr_b or b""
466 # END status handling
468 if status != 0:
469 errstr = read_all_from_possibly_closed_stream(p_stderr)
470 _logger.debug("AutoInterrupt wait stderr: %r" % (errstr,))
471 raise GitCommandError(remove_password_if_present(self.args), status, errstr)
472 return status
475_AutoInterrupt.__name__ = "AutoInterrupt"
476_AutoInterrupt.__qualname__ = "Git.AutoInterrupt"
479class _CatFileContentStream:
480 """Object representing a sized read-only stream returning the contents of
481 an object.
483 This behaves like a stream, but counts the data read and simulates an empty stream
484 once our sized content region is empty.
486 If not all data are read to the end of the object's lifetime, we read the rest to
487 ensure the underlying stream continues to work.
488 """
490 __slots__ = ("_stream", "_nbr", "_size")
492 def __init__(self, size: int, stream: IO[bytes]) -> None:
493 self._stream = stream
494 self._size = size
495 self._nbr = 0 # Number of bytes read.
497 # Special case: If the object is empty, has null bytes, get the final
498 # newline right away.
499 if size == 0:
500 stream.read(1)
501 # END handle empty streams
503 def read(self, size: int = -1) -> bytes:
504 bytes_left = self._size - self._nbr
505 if bytes_left == 0:
506 return b""
507 if size > -1:
508 # Ensure we don't try to read past our limit.
509 size = min(bytes_left, size)
510 else:
511 # They try to read all, make sure it's not more than what remains.
512 size = bytes_left
513 # END check early depletion
514 data = self._stream.read(size)
515 self._nbr += len(data)
517 # Check for depletion, read our final byte to make the stream usable by
518 # others.
519 if self._size - self._nbr == 0:
520 self._stream.read(1) # final newline
521 # END finish reading
522 return data
524 def readline(self, size: int = -1) -> bytes:
525 if self._nbr == self._size:
526 return b""
528 # Clamp size to lowest allowed value.
529 bytes_left = self._size - self._nbr
530 if size > -1:
531 size = min(bytes_left, size)
532 else:
533 size = bytes_left
534 # END handle size
536 data = self._stream.readline(size)
537 self._nbr += len(data)
539 # Handle final byte.
540 if self._size - self._nbr == 0:
541 self._stream.read(1)
542 # END finish reading
544 return data
546 def readlines(self, size: int = -1) -> List[bytes]:
547 if self._nbr == self._size:
548 return []
550 # Leave all additional logic to our readline method, we just check the size.
551 out = []
552 nbr = 0
553 while True:
554 line = self.readline()
555 if not line:
556 break
557 out.append(line)
558 if size > -1:
559 nbr += len(line)
560 if nbr > size:
561 break
562 # END handle size constraint
563 # END readline loop
564 return out
566 # skipcq: PYL-E0301
567 def __iter__(self) -> "Git.CatFileContentStream":
568 return self
570 def __next__(self) -> bytes:
571 line = self.readline()
572 if not line:
573 raise StopIteration
575 return line
577 next = __next__
579 def __del__(self) -> None:
580 bytes_left = self._size - self._nbr
581 if bytes_left:
582 # Read and discard - seeking is impossible within a stream.
583 # This includes any terminating newline.
584 self._stream.read(bytes_left + 1)
585 # END handle incomplete read
588_CatFileContentStream.__name__ = "CatFileContentStream"
589_CatFileContentStream.__qualname__ = "Git.CatFileContentStream"
592_USE_SHELL_DEFAULT_MESSAGE = (
593 "Git.USE_SHELL is deprecated, because only its default value of False is safe. "
594 "It will be removed in a future release."
595)
597_USE_SHELL_DANGER_MESSAGE = (
598 "Setting Git.USE_SHELL to True is unsafe and insecure, as the effect of special "
599 "shell syntax cannot usually be accounted for. This can result in a command "
600 "injection vulnerability and arbitrary code execution. Git.USE_SHELL is deprecated "
601 "and will be removed in a future release."
602)
605def _warn_use_shell(*, extra_danger: bool) -> None:
606 warnings.warn(
607 _USE_SHELL_DANGER_MESSAGE if extra_danger else _USE_SHELL_DEFAULT_MESSAGE,
608 DeprecationWarning,
609 stacklevel=3,
610 )
613class _GitMeta(type):
614 """Metaclass for :class:`Git`.
616 This helps issue :class:`DeprecationWarning` if :attr:`Git.USE_SHELL` is used.
617 """
619 def __getattribute(cls, name: str) -> Any:
620 if name == "USE_SHELL":
621 _warn_use_shell(extra_danger=False)
622 return super().__getattribute__(name)
624 def __setattr(cls, name: str, value: Any) -> Any:
625 if name == "USE_SHELL":
626 _warn_use_shell(extra_danger=value)
627 super().__setattr__(name, value)
629 if not TYPE_CHECKING:
630 # To preserve static checking for undefined/misspelled attributes while letting
631 # the methods' bodies be type-checked, these are defined as non-special methods,
632 # then bound to special names out of view of static type checkers. (The original
633 # names invoke name mangling (leading "__") to avoid confusion in other scopes.)
634 __getattribute__ = __getattribute
635 __setattr__ = __setattr
638GitMeta = _GitMeta
639"""Alias of :class:`Git`'s metaclass, whether it is :class:`type` or a custom metaclass.
641Whether the :class:`Git` class has the default :class:`type` as its metaclass or uses a
642custom metaclass is not documented and may change at any time. This statically checkable
643metaclass alias is equivalent at runtime to ``type(Git)``. This should almost never be
644used. Code that benefits from it is likely to be remain brittle even if it is used.
646In view of the :class:`Git` class's intended use and :class:`Git` objects' dynamic
647callable attributes representing git subcommands, it rarely makes sense to inherit from
648:class:`Git` at all. Using :class:`Git` in multiple inheritance can be especially tricky
649to do correctly. Attempting uses of :class:`Git` where its metaclass is relevant, such
650as when a sibling class has an unrelated metaclass and a shared lower bound metaclass
651might have to be introduced to solve a metaclass conflict, is not recommended.
653:note:
654 The correct static type of the :class:`Git` class itself, and any subclasses, is
655 ``Type[Git]``. (This can be written as ``type[Git]`` in Python 3.9 later.)
657 :class:`GitMeta` should never be used in any annotation where ``Type[Git]`` is
658 intended or otherwise possible to use. This alias is truly only for very rare and
659 inherently precarious situations where it is necessary to deal with the metaclass
660 explicitly.
661"""
664class Git(metaclass=_GitMeta):
665 """The Git class manages communication with the Git binary.
667 It provides a convenient interface to calling the Git binary, such as in::
669 g = Git( git_dir )
670 g.init() # calls 'git init' program
671 rval = g.ls_files() # calls 'git ls-files' program
673 Debugging:
675 * Set the :envvar:`GIT_PYTHON_TRACE` environment variable to print each invocation
676 of the command to stdout.
677 * Set its value to ``full`` to see details about the returned values.
678 """
680 __slots__ = (
681 "_working_dir",
682 "cat_file_all",
683 "cat_file_header",
684 "_version_info",
685 "_version_info_token",
686 "_git_options",
687 "_persistent_git_options",
688 "_environment",
689 )
691 _excluded_ = (
692 "cat_file_all",
693 "cat_file_header",
694 "_version_info",
695 "_version_info_token",
696 )
698 # Match Git's leading transport selector, including an empty helper name.
699 # Git also selects the command-executing ext helper for an ext:// URL.
700 re_unsafe_protocol = re.compile(r"([A-Za-z0-9][A-Za-z0-9+.-]*|)::|ext://")
702 unsafe_git_ls_remote_options = [
703 # This option allows arbitrary command execution in git-ls-remote.
704 "--upload-pack",
705 "--exec",
706 ]
708 unsafe_git_pathspec_from_file_options = [
709 # Reads pathspecs from a caller-controlled file. Some commands include an
710 # unmatched pathspec in their error output, which can disclose the file.
711 "--pathspec-from-file",
712 ]
714 def __getstate__(self) -> Dict[str, Any]:
715 return slots_to_dict(self, exclude=self._excluded_)
717 def __setstate__(self, d: Dict[str, Any]) -> None:
718 dict_to_slots_and__excluded_are_none(self, d, excluded=self._excluded_)
720 # CONFIGURATION
722 git_exec_name = "git"
723 """Default git command that should work on Linux, Windows, and other systems."""
725 GIT_PYTHON_TRACE = os.environ.get("GIT_PYTHON_TRACE", False)
726 """Enables debugging of GitPython's git commands."""
728 USE_SHELL: bool = False
729 """Deprecated. If set to ``True``, a shell will be used when executing git commands.
731 Code that uses ``USE_SHELL = True`` or that passes ``shell=True`` to any GitPython
732 functions should be updated to use the default value of ``False`` instead. ``True``
733 is unsafe unless the effect of syntax treated specially by the shell is fully
734 considered and accounted for, which is not possible under most circumstances. As
735 detailed below, it is also no longer needed, even where it had been in the past.
737 It is in many if not most cases a command injection vulnerability for an application
738 to set :attr:`USE_SHELL` to ``True``. Any attacker who can cause a specially crafted
739 fragment of text to make its way into any part of any argument to any git command
740 (including paths, branch names, etc.) can cause the shell to read and write
741 arbitrary files and execute arbitrary commands. Innocent input may also accidentally
742 contain special shell syntax, leading to inadvertent malfunctions.
744 In addition, how a value of ``True`` interacts with some aspects of GitPython's
745 operation is not precisely specified and may change without warning, even before
746 GitPython 4.0.0 when :attr:`USE_SHELL` may be removed. This includes:
748 * Whether or how GitPython automatically customizes the shell environment.
750 * Whether, outside of Windows (where :class:`subprocess.Popen` supports lists of
751 separate arguments even when ``shell=True``), this can be used with any GitPython
752 functionality other than direct calls to the :meth:`execute` method.
754 * Whether any GitPython feature that runs git commands ever attempts to partially
755 sanitize data a shell may treat specially. Currently this is not done.
757 Prior to GitPython 2.0.8, this had a narrow purpose in suppressing console windows
758 in graphical Windows applications. In 2.0.8 and higher, it provides no benefit, as
759 GitPython solves that problem more robustly and safely by using the
760 ``CREATE_NO_WINDOW`` process creation flag on Windows.
762 Because Windows path search differs subtly based on whether a shell is used, in rare
763 cases changing this from ``True`` to ``False`` may keep an unusual git "executable",
764 such as a batch file, from being found. To fix this, set the command name or full
765 path in the :envvar:`GIT_PYTHON_GIT_EXECUTABLE` environment variable or pass the
766 full path to :func:`git.refresh` (or invoke the script using a ``.exe`` shim).
768 Further reading:
770 * :meth:`Git.execute` (on the ``shell`` parameter).
771 * https://github.com/gitpython-developers/GitPython/commit/0d9390866f9ce42870d3116094cd49e0019a970a
772 * https://learn.microsoft.com/en-us/windows/win32/procthread/process-creation-flags
773 * https://github.com/python/cpython/issues/91558#issuecomment-1100942950
774 * https://learn.microsoft.com/en-us/windows/win32/api/processthreadsapi/nf-processthreadsapi-createprocessw
775 """
777 _git_exec_env_var = "GIT_PYTHON_GIT_EXECUTABLE"
778 _refresh_env_var = "GIT_PYTHON_REFRESH"
780 GIT_PYTHON_GIT_EXECUTABLE = None
781 """Provide the full path to the git executable. Otherwise it assumes git is in the
782 executable search path.
784 :note:
785 The git executable is actually found during the refresh step in the top level
786 ``__init__``. It can also be changed by explicitly calling :func:`git.refresh`.
787 """
789 _refresh_token = object() # Since None would match an initial _version_info_token.
791 @classmethod
792 def refresh(cls, path: Union[None, PathLike] = None) -> bool:
793 """Update information about the git executable :class:`Git` objects will use.
795 Called by the :func:`git.refresh` function in the top level ``__init__``.
797 :param path:
798 Optional path to the git executable. If not absolute, it is resolved
799 immediately, relative to the current directory. (See note below.)
801 :note:
802 The top-level :func:`git.refresh` should be preferred because it calls this
803 method and may also update other state accordingly.
805 :note:
806 There are three different ways to specify the command that refreshing causes
807 to be used for git:
809 1. Pass no `path` argument and do not set the
810 :envvar:`GIT_PYTHON_GIT_EXECUTABLE` environment variable. The command
811 name ``git`` is used. It is looked up in a path search by the system, in
812 each command run (roughly similar to how git is found when running
813 ``git`` commands manually). This is usually the desired behavior.
815 2. Pass no `path` argument but set the :envvar:`GIT_PYTHON_GIT_EXECUTABLE`
816 environment variable. The command given as the value of that variable is
817 used. This may be a simple command or an arbitrary path. It is looked up
818 in each command run. Setting :envvar:`GIT_PYTHON_GIT_EXECUTABLE` to
819 ``git`` has the same effect as not setting it.
821 3. Pass a `path` argument. This path, if not absolute, is immediately
822 resolved, relative to the current directory. This resolution occurs at
823 the time of the refresh. When git commands are run, they are run using
824 that previously resolved path. If a `path` argument is passed, the
825 :envvar:`GIT_PYTHON_GIT_EXECUTABLE` environment variable is not
826 consulted.
828 :note:
829 Refreshing always sets the :attr:`Git.GIT_PYTHON_GIT_EXECUTABLE` class
830 attribute, which can be read on the :class:`Git` class or any of its
831 instances to check what command is used to run git. This attribute should
832 not be confused with the related :envvar:`GIT_PYTHON_GIT_EXECUTABLE`
833 environment variable. The class attribute is set no matter how refreshing is
834 performed.
835 """
836 # Discern which path to refresh with.
837 if path is not None:
838 new_git = os.path.expanduser(path)
839 new_git = os.path.abspath(new_git)
840 else:
841 new_git = os.environ.get(cls._git_exec_env_var, cls.git_exec_name)
843 # Keep track of the old and new git executable path.
844 old_git = cls.GIT_PYTHON_GIT_EXECUTABLE
845 old_refresh_token = cls._refresh_token
846 cls.GIT_PYTHON_GIT_EXECUTABLE = new_git
847 cls._refresh_token = object()
849 # Test if the new git executable path is valid. A GitCommandNotFound error is
850 # raised by us. A PermissionError is raised if the git executable cannot be
851 # executed for whatever reason.
852 has_git = False
853 try:
854 cls().version()
855 has_git = True
856 except (GitCommandNotFound, PermissionError):
857 pass
859 # Warn or raise exception if test failed.
860 if not has_git:
861 err = (
862 dedent(
863 """\
864 Bad git executable.
865 The git executable must be specified in one of the following ways:
866 - be included in your $PATH
867 - be set via $%s
868 - explicitly set via git.refresh(<full-path-to-git-executable>)
869 """
870 )
871 % cls._git_exec_env_var
872 )
874 # Revert to whatever the old_git was.
875 cls.GIT_PYTHON_GIT_EXECUTABLE = old_git
876 cls._refresh_token = old_refresh_token
878 if old_git is None:
879 # On the first refresh (when GIT_PYTHON_GIT_EXECUTABLE is None) we only
880 # are quiet, warn, or error depending on the GIT_PYTHON_REFRESH value.
882 # Determine what the user wants to happen during the initial refresh. We
883 # expect GIT_PYTHON_REFRESH to either be unset or be one of the
884 # following values:
885 #
886 # 0|q|quiet|s|silence|silent|n|none
887 # 1|w|warn|warning|l|log
888 # 2|r|raise|e|error|exception
890 mode = os.environ.get(cls._refresh_env_var, "raise").lower()
892 quiet = ["quiet", "q", "silence", "s", "silent", "none", "n", "0"]
893 warn = ["warn", "w", "warning", "log", "l", "1"]
894 error = ["error", "e", "exception", "raise", "r", "2"]
896 if mode in quiet:
897 pass
898 elif mode in warn or mode in error:
899 err = dedent(
900 """\
901 %s
902 All git commands will error until this is rectified.
904 This initial message can be silenced or aggravated in the future by setting the
905 $%s environment variable. Use one of the following values:
906 - %s: for no message or exception
907 - %s: for a warning message (logging level CRITICAL, displayed by default)
908 - %s: for a raised exception
910 Example:
911 export %s=%s
912 """
913 ) % (
914 err,
915 cls._refresh_env_var,
916 "|".join(quiet),
917 "|".join(warn),
918 "|".join(error),
919 cls._refresh_env_var,
920 quiet[0],
921 )
923 if mode in warn:
924 _logger.critical(err)
925 else:
926 raise ImportError(err)
927 else:
928 err = dedent(
929 """\
930 %s environment variable has been set but it has been set with an invalid value.
932 Use only the following values:
933 - %s: for no message or exception
934 - %s: for a warning message (logging level CRITICAL, displayed by default)
935 - %s: for a raised exception
936 """
937 ) % (
938 cls._refresh_env_var,
939 "|".join(quiet),
940 "|".join(warn),
941 "|".join(error),
942 )
943 raise ImportError(err)
945 # We get here if this was the initial refresh and the refresh mode was
946 # not error. Go ahead and set the GIT_PYTHON_GIT_EXECUTABLE such that we
947 # discern the difference between the first refresh at import time
948 # and subsequent calls to git.refresh or this refresh method.
949 cls.GIT_PYTHON_GIT_EXECUTABLE = cls.git_exec_name
950 else:
951 # After the first refresh (when GIT_PYTHON_GIT_EXECUTABLE is no longer
952 # None) we raise an exception.
953 raise GitCommandNotFound(new_git, err)
955 return has_git
957 @classmethod
958 def is_cygwin(cls) -> bool:
959 return is_cygwin_git(cls.GIT_PYTHON_GIT_EXECUTABLE)
961 @overload
962 @classmethod
963 def polish_url(cls, url: str, is_cygwin: Literal[False] = ..., expand_vars: bool = ...) -> str: ...
965 @overload
966 @classmethod
967 def polish_url(cls, url: str, is_cygwin: Union[None, bool] = None, expand_vars: bool = True) -> str: ...
969 @classmethod
970 def polish_url(cls, url: str, is_cygwin: Union[None, bool] = None, expand_vars: bool = True) -> PathLike:
971 """Remove any backslashes from URLs to be written in config files.
973 Windows might create config files containing paths with backslashes, but git
974 stops liking them as it will escape the backslashes. Hence we undo the escaping
975 just to be sure.
977 :param expand_vars:
978 Expand environment variables and an initial ``~``. Disable this for values
979 obtained from an untrusted source, such as remote URLs.
980 """
981 if is_cygwin is None:
982 is_cygwin = cls.is_cygwin()
984 if is_cygwin:
985 url = cygpath(url, expand_vars=expand_vars)
986 else:
987 if expand_vars:
988 url = os.path.expandvars(url)
989 if url.startswith("~"):
990 url = os.path.expanduser(url)
991 url = url.replace("\\\\", "\\").replace("\\", "/")
992 return url
994 @classmethod
995 def check_unsafe_protocols(cls, url: str) -> None:
996 """Check for unsafe protocols.
998 Apart from the usual protocols (http, git, ssh), Git allows "remote helpers"
999 that have the form ``<transport>::<address>``. One of these helpers (``ext::``)
1000 can be used to invoke any arbitrary command. Git also selects that helper
1001 for ``ext://`` URLs and interprets the URL as a command path.
1003 See:
1005 - https://git-scm.com/docs/gitremote-helpers
1006 - https://git-scm.com/docs/git-remote-ext
1007 """
1008 match = cls.re_unsafe_protocol.match(url)
1009 if match:
1010 protocol = match.group(0)
1011 raise UnsafeProtocolError(
1012 f"The `{protocol}` protocol looks suspicious, use `allow_unsafe_protocols=True` to allow it."
1013 )
1015 def _check_unsafe_protocols_in_args(self, args: Sequence[Any], kwargs: Mapping[str, Any]) -> None:
1016 """Check positional operands and standalone values in rendered command options.
1018 A short flag's split value can become the repository operand before ``--``.
1019 """
1020 for arg in self._unpack_args(args) + self.transform_kwargs(**kwargs):
1021 self.check_unsafe_protocols(arg)
1023 @classmethod
1024 def _canonicalize_option_name(cls, option: str) -> str:
1025 """Return the option name used for unsafe-option checks.
1027 Examples:
1028 ``"--upload-pack=/tmp/helper"`` -> ``"upload-pack"``
1029 ``"upload_pack"`` -> ``"upload-pack"``
1030 ``"--config core.filemode=false"`` -> ``"config"``
1031 """
1032 option_name = option.lstrip("-").split("=", 1)[0]
1033 option_tokens = option_name.split(None, 1)
1034 if not option_tokens:
1035 return ""
1036 return dashify(option_tokens[0])
1038 @classmethod
1039 def check_unsafe_options(
1040 cls, options: List[str], unsafe_options: List[str], clusterable_short_options: str = "46flnqsv"
1041 ) -> None:
1042 """Raise :class:`~git.exc.UnsafeOptionError` for blocked option spellings.
1044 In addition to exact matches, this rejects abbreviated long options accepted
1045 by Git (for example, ``--upl`` for ``--upload-pack``) and unsafe short options
1046 whose values are joined to the same token, including after clusterable flags
1047 (for example, ``-uVALUE`` and ``-fuVALUE``).
1049 A list containing only bare names is treated as normalized keyword arguments,
1050 so multi-character names such as ``upload_p`` are checked as long-option
1051 abbreviations. If any item starts with ``-``, the list is treated as tokenized
1052 command-line input: bare items can be option values and are not checked as
1053 abbreviations. Thus ``["--origin", "upload"]`` is allowed. Single-dash options
1054 use short-option parsing rather than broad prefix matching, preserving safe
1055 attached values such as ``-oupstream`` and ``-bcurrent``.
1057 Some options passed to ``git <command>`` can execute arbitrary commands and
1058 are therefore blocked by default unless the caller explicitly allows them.
1059 """
1060 # Options can be of the form `foo`, `--foo`, `--foo bar`, or `--foo=bar`.
1061 # Git accepts any unambiguous prefix of a long option, so an abbreviated
1062 # spelling such as `--upl` for `--upload-pack` must be rejected too. An
1063 # option is unsafe if its canonical name is a prefix of any blocked
1064 # option's canonical name. Only long options and multi-character kwargs
1065 # can be abbreviations; single-character short options remain exact-match
1066 # only.
1067 canonical_unsafe_options = {cls._canonicalize_option_name(option): option for option in unsafe_options}
1068 unsafe_short_options = {
1069 canonical: option
1070 for canonical, option in canonical_unsafe_options.items()
1071 if option.startswith("-") and not option.startswith("--") and len(canonical) == 1
1072 }
1073 # These value-less Git flags can be clustered before another short option
1074 # (for example, ``-fuVALUE``). Stop at any other character because it may
1075 # begin an attached value, as ``o`` does in the safe option ``-oupstream``.
1076 clusterable_short_options_set = frozenset(clusterable_short_options)
1077 options_are_kwargs = all(not option.startswith("-") for option in options)
1078 for option in options:
1079 candidate = cls._canonicalize_option_name(option)
1080 if not candidate:
1081 continue
1082 unsafe_option = canonical_unsafe_options.get(candidate)
1083 if unsafe_option is not None:
1084 raise UnsafeOptionError(f"{unsafe_option} is not allowed, use `allow_unsafe_options=True` to allow it.")
1085 option_token = option.split("=", 1)[0].split(None, 1)[0]
1086 if option_token.startswith("-") and not option_token.startswith("--"):
1087 for option_char in option_token[1:]:
1088 unsafe_option = unsafe_short_options.get(option_char)
1089 if unsafe_option is not None:
1090 raise UnsafeOptionError(
1091 f"{unsafe_option} is not allowed, use `allow_unsafe_options=True` to allow it."
1092 )
1093 if option_char not in clusterable_short_options_set:
1094 break
1095 if not (option.startswith("--") or (options_are_kwargs and len(candidate) > 1)):
1096 continue
1097 for canonical, unsafe_option in canonical_unsafe_options.items():
1098 if canonical.startswith(candidate):
1099 raise UnsafeOptionError(
1100 f"{unsafe_option} is not allowed, use `allow_unsafe_options=True` to allow it."
1101 )
1103 @classmethod
1104 def _option_candidates(cls, args: Sequence[Any] = (), kwargs: Optional[Mapping[str, Any]] = None) -> List[str]:
1105 """Collect possible option spellings before command-line transformation."""
1106 options = [
1107 option for option in cls._unpack_args([arg for arg in args if arg is not None]) if option.startswith("-")
1108 ]
1109 if kwargs:
1110 split_single_char_options = kwargs.get("split_single_char_options", True)
1111 for key, value in kwargs.items():
1112 values = value if isinstance(value, (list, tuple)) else (value,)
1113 if any(value is True or (value is not False and value is not None) for value in values):
1114 key = str(key)
1115 if len(key) != 1:
1116 options.append(f"--{dashify(key)}")
1117 elif split_single_char_options:
1118 options.append(f"-{key}")
1119 options.extend(
1120 str(value)
1121 for value in values
1122 if value is not True and value not in (False, None) and str(value).startswith("-")
1123 )
1124 else:
1125 options.extend(
1126 f"-{key}" if value is True else f"-{key}{value}"
1127 for value in values
1128 if value is True or (value is not False and value is not None)
1129 )
1130 return options
1132 AutoInterrupt: TypeAlias = _AutoInterrupt
1134 CatFileContentStream: TypeAlias = _CatFileContentStream
1136 def __init__(self, working_dir: Union[None, PathLike] = None) -> None:
1137 """Initialize this instance with:
1139 :param working_dir:
1140 Git directory we should work in. If ``None``, we always work in the current
1141 directory as returned by :func:`os.getcwd`.
1142 This is meant to be the working tree directory if available, or the
1143 ``.git`` directory in case of bare repositories.
1144 """
1145 super().__init__()
1146 self._working_dir = expand_path(working_dir)
1147 self._git_options: Union[List[str], Tuple[str, ...]] = ()
1148 self._persistent_git_options: List[str] = []
1150 # Extra environment variables to pass to git commands
1151 self._environment: Dict[str, str] = {}
1153 # Cached version slots
1154 self._version_info: Union[Tuple[int, ...], None] = None
1155 self._version_info_token: object = None
1157 # Cached command slots
1158 self.cat_file_header: Union[None, TBD] = None
1159 self.cat_file_all: Union[None, TBD] = None
1161 def __getattribute__(self, name: str) -> Any:
1162 if name == "USE_SHELL":
1163 _warn_use_shell(extra_danger=False)
1164 return super().__getattribute__(name)
1166 def __getattr__(self, name: str) -> Any:
1167 """A convenience method as it allows to call the command as if it was an object.
1169 :return:
1170 Callable object that will execute call :meth:`_call_process` with your
1171 arguments.
1172 """
1173 if name.startswith("_"):
1174 return super().__getattribute__(name)
1175 return lambda *args, **kwargs: self._call_process(name, *args, **kwargs)
1177 def set_persistent_git_options(self, **kwargs: Any) -> None:
1178 """Specify command line options to the git executable for subsequent
1179 subcommand calls.
1181 :param kwargs:
1182 A dict of keyword arguments.
1183 These arguments are passed as in :meth:`_call_process`, but will be passed
1184 to the git command rather than the subcommand.
1185 """
1187 self._persistent_git_options = self.transform_kwargs(split_single_char_options=True, **kwargs)
1189 def ls_remote(
1190 self,
1191 *args: Any,
1192 allow_unsafe_options: bool = False,
1193 allow_unsafe_protocols: bool = False,
1194 **kwargs: Any,
1195 ) -> Union[str, bytes, Tuple[int, Union[str, bytes], str], "Git.AutoInterrupt"]:
1196 """List references in a remote repository.
1198 :param allow_unsafe_options:
1199 Allow unsafe options, like ``--upload-pack`` or ``--exec``.
1201 :param allow_unsafe_protocols:
1202 Allow unsafe protocols to be used, like ``ext``. Positional arguments
1203 and split short-option values are checked.
1204 """
1205 if not allow_unsafe_options:
1206 candidate_options = self._option_candidates(args, kwargs)
1207 Git.check_unsafe_options(options=candidate_options, unsafe_options=self.unsafe_git_ls_remote_options)
1208 if not allow_unsafe_protocols:
1209 self._check_unsafe_protocols_in_args(args, kwargs)
1210 return self._call_process("ls_remote", *args, **kwargs)
1212 @property
1213 def working_dir(self) -> Union[None, PathLike]:
1214 """:return: Git directory we are working on"""
1215 return self._working_dir
1217 @property
1218 def version_info(self) -> Tuple[int, ...]:
1219 """
1220 :return: Tuple with integers representing the major, minor and additional
1221 version numbers as parsed from :manpage:`git-version(1)`. Up to four fields
1222 are used.
1224 This value is generated on demand and is cached.
1225 """
1226 # Refreshing is global, but version_info caching is per-instance.
1227 refresh_token = self._refresh_token # Copy token in case of concurrent refresh.
1229 # Use the cached version if obtained after the most recent refresh.
1230 if self._version_info_token is refresh_token:
1231 assert self._version_info is not None, "Bug: corrupted token-check state"
1232 return self._version_info
1234 # Run "git version" and parse it.
1235 process_version = self._call_process("version")
1236 version_string = process_version.split(" ")[2]
1237 version_fields = version_string.split(".")[:4]
1238 leading_numeric_fields = itertools.takewhile(str.isdigit, version_fields)
1239 self._version_info = tuple(map(int, leading_numeric_fields))
1241 # This value will be considered valid until the next refresh.
1242 self._version_info_token = refresh_token
1243 return self._version_info
1245 @overload
1246 def execute(
1247 self,
1248 command: Union[str, Sequence[Any]],
1249 istream: Union[None, int, BinaryIO] = None,
1250 *,
1251 as_process: Literal[True],
1252 **subprocess_kwargs: Any,
1253 ) -> "AutoInterrupt": ...
1255 @overload
1256 def execute(
1257 self,
1258 command: Union[str, Sequence[Any]],
1259 istream: Union[None, int, BinaryIO] = None,
1260 *,
1261 as_process: Literal[False] = False,
1262 with_extended_output: Literal[False] = False,
1263 stdout_as_string: Literal[True] = True,
1264 with_stdout: Literal[True] = True,
1265 **subprocess_kwargs: Any,
1266 ) -> str: ...
1268 @overload
1269 def execute(
1270 self,
1271 command: Union[str, Sequence[Any]],
1272 istream: Union[None, int, BinaryIO] = None,
1273 *,
1274 as_process: Literal[False] = False,
1275 with_extended_output: Literal[False] = False,
1276 stdout_as_string: Literal[False],
1277 universal_newlines: Literal[False] = False,
1278 with_stdout: Literal[True] = True,
1279 **subprocess_kwargs: Any,
1280 ) -> bytes: ...
1282 @overload
1283 def execute(
1284 self,
1285 command: Union[str, Sequence[Any]],
1286 istream: Union[None, int, BinaryIO] = None,
1287 *,
1288 as_process: Literal[False] = False,
1289 with_extended_output: Literal[True],
1290 stdout_as_string: Literal[True] = True,
1291 with_stdout: Literal[True] = True,
1292 **subprocess_kwargs: Any,
1293 ) -> Tuple[int, str, str]: ...
1295 @overload
1296 def execute(
1297 self,
1298 command: Union[str, Sequence[Any]],
1299 istream: Union[None, int, BinaryIO] = None,
1300 *,
1301 as_process: Literal[False] = False,
1302 with_extended_output: Literal[True],
1303 stdout_as_string: Literal[False],
1304 universal_newlines: Literal[False] = False,
1305 with_stdout: Literal[True] = True,
1306 **subprocess_kwargs: Any,
1307 ) -> Tuple[int, bytes, str]: ...
1309 @overload
1310 def execute(
1311 self,
1312 command: Union[str, Sequence[Any]],
1313 istream: Union[None, int, BinaryIO] = None,
1314 *,
1315 as_process: Literal[False] = False,
1316 with_extended_output: Literal[True],
1317 **subprocess_kwargs: Any,
1318 ) -> Tuple[int, Union[str, bytes, None], str]: ...
1320 @overload
1321 def execute(
1322 self,
1323 command: Union[str, Sequence[Any]],
1324 istream: Union[None, int, BinaryIO] = None,
1325 *,
1326 as_process: Literal[False] = False,
1327 with_extended_output: Literal[False] = False,
1328 **subprocess_kwargs: Any,
1329 ) -> Union[str, bytes, None]: ...
1331 @overload
1332 def execute(
1333 self,
1334 command: Union[str, Sequence[Any]],
1335 istream: Union[None, int, BinaryIO] = None,
1336 with_extended_output: bool = False,
1337 with_exceptions: bool = True,
1338 as_process: bool = False,
1339 output_stream: Union[None, BinaryIO] = None,
1340 stdout_as_string: bool = True,
1341 kill_after_timeout: Union[None, float] = None,
1342 with_stdout: bool = True,
1343 universal_newlines: bool = False,
1344 shell: Union[None, bool] = None,
1345 env: Union[None, Mapping[str, str]] = None,
1346 max_chunk_size: int = io.DEFAULT_BUFFER_SIZE,
1347 strip_newline_in_stdout: bool = True,
1348 **subprocess_kwargs: Any,
1349 ) -> Union[None, str, bytes, Tuple[int, Union[str, bytes, None], str], AutoInterrupt]: ...
1351 def execute(
1352 self,
1353 command: Union[str, Sequence[Any]],
1354 istream: Union[None, int, BinaryIO] = None,
1355 with_extended_output: bool = False,
1356 with_exceptions: bool = True,
1357 as_process: bool = False,
1358 output_stream: Union[None, BinaryIO] = None,
1359 stdout_as_string: bool = True,
1360 kill_after_timeout: Union[None, float] = None,
1361 with_stdout: bool = True,
1362 universal_newlines: bool = False,
1363 shell: Union[None, bool] = None,
1364 env: Union[None, Mapping[str, str]] = None,
1365 max_chunk_size: int = io.DEFAULT_BUFFER_SIZE,
1366 strip_newline_in_stdout: bool = True,
1367 **subprocess_kwargs: Any,
1368 ) -> Union[None, str, bytes, Tuple[int, Union[str, bytes, None], str], AutoInterrupt]:
1369 R"""Handle executing the command, and consume and return the returned
1370 information (stdout).
1372 :param command:
1373 The command to execute. A sequence of program arguments is recommended.
1374 A string is also accepted, but its meaning is strongly platform-dependent.
1376 By default, a shell is not used. On Unix-like systems, a string is the whole
1377 program name (so ``"git log -n 1"`` raises :class:`GitCommandNotFound`). On
1378 Windows, the program parses the arguments itself, so multi-word strings can
1379 work but are not portable.
1381 Avoid ``shell=True`` (and :attr:`Git.USE_SHELL`): this runs the command in
1382 a shell, which is generally unsafe. The shell interprets metacharacters
1383 such as ``;``, ``|``, ``&``, ``$(...)``, ``$VAR``, ``%VAR%``, and ``^``
1384 (depending on the platform) as syntax. Any untrusted text in the command
1385 can then execute arbitrary OS commands. See :attr:`Git.USE_SHELL`.
1387 Producing a sequence automatically by :func:`shlex.split` and passing it
1388 as the command is far safer than ``shell=True``. But :func:`shlex.split`
1389 parses POSIX shell syntax on all systems, and the result is still unsafe
1390 for anything but *fixed, fully trusted* strings. Do not use it on strings
1391 built by interpolating values: whitespace or quoting in an untrusted value
1392 can still inject arguments. For input derived in any way from untrusted
1393 data, build the argument sequence yourself, while ensuring each argument
1394 is fully sanitized.
1396 :param istream:
1397 Standard input filehandle passed to :class:`subprocess.Popen`.
1399 :param with_extended_output:
1400 Whether to return a (status, stdout, stderr) tuple.
1402 :param with_exceptions:
1403 Whether to raise an exception when git returns a non-zero status.
1405 :param as_process:
1406 Whether to return the created process instance directly from which
1407 streams can be read on demand. This will render `with_extended_output`
1408 and `with_exceptions` ineffective - the caller will have to deal with
1409 the details. It is important to note that the process will be placed
1410 into an :class:`AutoInterrupt` wrapper that will interrupt the process
1411 once it goes out of scope. If you use the command in iterators, you
1412 should pass the whole process instance instead of a single stream.
1414 :param output_stream:
1415 If set to a file-like object, data produced by the git command will be
1416 copied to the given stream instead of being returned as a string.
1417 This feature only has any effect if `as_process` is ``False``.
1419 :param stdout_as_string:
1420 If ``False``, the command's standard output will be bytes. Otherwise, it
1421 will be decoded into a string using the default encoding (usually UTF-8).
1422 The latter can fail, if the output contains binary data.
1424 :param kill_after_timeout:
1425 Specifies a timeout in seconds for the git command, after which the process
1426 should be killed. This will have no effect if `as_process` is set to
1427 ``True``. It is set to ``None`` by default and will let the process run
1428 until the timeout is explicitly specified. Uses of this feature should be
1429 carefully considered, due to the following limitations:
1431 1. This feature is not supported at all on Windows.
1432 2. Enumerating child processes requires ``pgrep -P``, or a ``ps`` command
1433 supporting the POSIX ``-A`` and ``-o`` options if ``pgrep`` is not
1434 installed (``ps -ef`` on Cygwin). Effectiveness may vary on systems
1435 without these commands.
1436 3. Descendants are enumerated before signalling. Processes that detach
1437 or spawn after enumeration may not receive signals.
1438 4. `kill_after_timeout` uses ``SIGKILL``, which can have negative side
1439 effects on a repository. For example, stale locks in case of
1440 :manpage:`git-gc(1)` could render the repository incapable of accepting
1441 changes until the lock is manually removed.
1443 :param with_stdout:
1444 If ``True``, default ``True``, we open stdout on the created process.
1446 :param universal_newlines:
1447 If ``True``, pipes will be opened as text, and lines are split at all known
1448 line endings.
1450 :param shell:
1451 Whether to invoke commands through a shell
1452 (see :class:`Popen(..., shell=True) <subprocess.Popen>`).
1453 If this is not ``None``, it overrides :attr:`USE_SHELL`.
1455 Passing ``shell=True`` to this or any other GitPython function should be
1456 avoided, as it is unsafe under most circumstances. This is because it is
1457 typically not feasible to fully consider and account for the effect of shell
1458 expansions, especially when passing ``shell=True`` to other methods that
1459 forward it to :meth:`Git.execute`. Passing ``shell=True`` is also no longer
1460 needed (nor useful) to work around any known operating system specific
1461 issues.
1463 On Unix-like systems, when migrating away from passing string commands with
1464 ``shell=True``, :func:`shlex.split` may serve as a transitional step in rare
1465 cases, with extreme care. (Drop ``shell=True`` and pass the resulting
1466 sequence as the command.) See the `command` parameter above on the risks.
1468 :param env:
1469 A dictionary of environment variables to be passed to
1470 :class:`subprocess.Popen`.
1472 :param max_chunk_size:
1473 Maximum number of bytes in one chunk of data passed to the `output_stream`
1474 in one invocation of its ``write()`` method. If the given number is not
1475 positive then the default value is used.
1477 :param strip_newline_in_stdout:
1478 Whether to strip the trailing ``\n`` of the command stdout.
1480 :param subprocess_kwargs:
1481 Keyword arguments to be passed to :class:`subprocess.Popen`. Please note
1482 that some of the valid kwargs are already set by this method; the ones you
1483 specify may not be the same ones.
1485 :return:
1486 * str(output), if `extended_output` is ``False`` (Default)
1487 * tuple(int(status), str(stdout), str(stderr)),
1488 if `extended_output` is ``True``
1490 If `output_stream` is ``True``, the stdout value will be your output stream:
1492 * output_stream, if `extended_output` is ``False``
1493 * tuple(int(status), output_stream, str(stderr)),
1494 if `extended_output` is ``True``
1496 Note that git is executed with ``LC_MESSAGES="C"`` to ensure consistent
1497 output regardless of system language.
1499 :raise git.exc.GitCommandError:
1501 :note:
1502 If you add additional keyword arguments to the signature of this method, you
1503 must update the ``execute_kwargs`` variable housed in this module.
1504 """
1505 # Remove password for the command if present.
1506 redacted_command = remove_password_if_present(command)
1507 if self.GIT_PYTHON_TRACE and (self.GIT_PYTHON_TRACE != "full" or as_process):
1508 _logger.info(" ".join(redacted_command))
1510 # Allow the user to have the command executed in their working dir.
1511 try:
1512 cwd = self._working_dir or os.getcwd() # type: Optional[PathLike]
1513 if not os.access(str(cwd), os.X_OK):
1514 cwd = None
1515 except FileNotFoundError:
1516 cwd = None
1518 # Start the process.
1519 inline_env = env
1520 env = os.environ.copy()
1521 # Attempt to force all output to plain ASCII English, which is what some parsing
1522 # code may expect.
1523 # According to https://askubuntu.com/a/311796, we are setting LANGUAGE as well
1524 # just to be sure.
1525 env["LANGUAGE"] = "C"
1526 env["LC_ALL"] = "C"
1527 env.update(self._environment)
1528 if inline_env is not None:
1529 env.update(inline_env)
1531 if sys.platform == "win32":
1532 if kill_after_timeout is not None:
1533 raise GitCommandError(
1534 redacted_command,
1535 '"kill_after_timeout" feature is not supported on Windows.',
1536 )
1537 cmd_not_found_exception = OSError
1538 else:
1539 cmd_not_found_exception = FileNotFoundError
1540 # END handle
1542 stdout_sink = PIPE if with_stdout else getattr(subprocess, "DEVNULL", None) or open(os.devnull, "wb")
1543 if shell is None:
1544 # Get the value of USE_SHELL with no deprecation warning. Do this without
1545 # warnings.catch_warnings, to avoid a race condition with application code
1546 # configuring warnings. The value could be looked up in type(self).__dict__
1547 # or Git.__dict__, but those can break under some circumstances. This works
1548 # the same as self.USE_SHELL in more situations; see Git.__getattribute__.
1549 shell = super().__getattribute__("USE_SHELL")
1550 _logger.debug(
1551 "Popen(%s, cwd=%s, stdin=%s, shell=%s, universal_newlines=%s)",
1552 redacted_command,
1553 cwd,
1554 "<valid stream>" if istream else "None",
1555 shell,
1556 universal_newlines,
1557 )
1558 try:
1559 proc = safer_popen(
1560 command,
1561 env=env,
1562 cwd=cwd,
1563 bufsize=-1,
1564 stdin=(istream or DEVNULL),
1565 stderr=PIPE,
1566 stdout=stdout_sink,
1567 shell=shell,
1568 universal_newlines=universal_newlines,
1569 encoding=defenc if universal_newlines else None,
1570 **subprocess_kwargs,
1571 )
1572 except cmd_not_found_exception as err:
1573 raise GitCommandNotFound(redacted_command, err) from err
1574 else:
1575 # Replace with a typeguard for Popen[bytes]?
1576 proc.stdout = cast(BinaryIO, proc.stdout)
1577 proc.stderr = cast(BinaryIO, proc.stderr)
1579 if as_process:
1580 return self.AutoInterrupt(proc, command)
1582 watchdog: Optional[threading.Timer] = None
1583 kill_check: Optional[threading.Event] = None
1584 timeout_error: Optional[Callable[[], Union[str, bytes]]] = None
1586 if sys.platform != "win32" and kill_after_timeout is not None:
1587 # Help mypy figure out this is not None even when used inside communicate().
1588 timeout = kill_after_timeout
1590 def kill_process(pid: int) -> None:
1591 if _kill_process(pid):
1592 assert kill_check is not None
1593 kill_check.set()
1595 def make_timeout_error() -> Union[str, bytes]:
1596 err = f'Timeout: the command "{" ".join(redacted_command)}" did not complete in {timeout:g} secs.'
1597 return err if universal_newlines else err.encode(defenc)
1599 def communicate() -> Tuple[Union[str, bytes, None], Union[str, bytes, None]]:
1600 assert watchdog is not None
1601 assert kill_check is not None
1602 watchdog.start()
1603 out, err = proc.communicate()
1604 watchdog.cancel()
1605 if kill_check.is_set():
1606 err = make_timeout_error()
1607 return out, err
1609 # END helpers
1611 kill_check = threading.Event()
1612 watchdog = threading.Timer(timeout, kill_process, args=(proc.pid,))
1613 timeout_error = make_timeout_error
1614 else:
1615 communicate = proc.communicate
1617 # Wait for the process to return.
1618 status = 0
1619 stdout_value: Union[str, bytes, None] = b""
1620 stderr_value: Union[str, bytes, None] = b""
1621 newline = "\n" if universal_newlines else b"\n"
1622 try:
1623 if output_stream is None:
1624 stdout_value, stderr_value = communicate()
1625 # Strip trailing "\n".
1626 if stdout_value is not None and stdout_value.endswith(newline) and strip_newline_in_stdout: # type: ignore[arg-type]
1627 stdout_value = stdout_value[:-1]
1628 if stderr_value is not None and stderr_value.endswith(newline): # type: ignore[arg-type]
1629 stderr_value = stderr_value[:-1]
1631 status = proc.returncode
1632 else:
1633 max_chunk_size = max_chunk_size if max_chunk_size and max_chunk_size > 0 else io.DEFAULT_BUFFER_SIZE
1634 if watchdog is not None:
1635 watchdog.start()
1636 try:
1637 if proc.stdout is not None:
1638 stream_copy(proc.stdout, output_stream, max_chunk_size)
1639 stdout_value = proc.stdout.read()
1640 if proc.stderr is not None:
1641 stderr_value = proc.stderr.read()
1642 status = proc.wait()
1643 finally:
1644 if watchdog is not None:
1645 watchdog.cancel()
1646 # Strip trailing "\n".
1647 if stderr_value is not None and stderr_value.endswith(newline): # type: ignore[arg-type]
1648 stderr_value = stderr_value[:-1]
1649 if kill_check is not None and kill_check.is_set():
1650 assert timeout_error is not None
1651 stderr_value = timeout_error()
1652 # END stdout handling
1653 finally:
1654 if proc.stdout is not None:
1655 proc.stdout.close()
1656 if proc.stderr is not None:
1657 proc.stderr.close()
1659 if self.GIT_PYTHON_TRACE == "full":
1660 cmdstr = " ".join(redacted_command)
1662 def as_text(stdout_value: Union[bytes, str, None]) -> str:
1663 return not output_stream and safe_decode(stdout_value) or "<OUTPUT_STREAM>"
1665 # END as_text
1667 if stderr_value:
1668 _logger.info(
1669 "%s -> %d; stdout: '%s'; stderr: '%s'",
1670 cmdstr,
1671 status,
1672 as_text(stdout_value),
1673 safe_decode(stderr_value),
1674 )
1675 elif stdout_value:
1676 _logger.info("%s -> %d; stdout: '%s'", cmdstr, status, as_text(stdout_value))
1677 else:
1678 _logger.info("%s -> %d", cmdstr, status)
1679 # END handle debug printing
1681 if with_exceptions and status != 0:
1682 raise GitCommandError(redacted_command, status, stderr_value, stdout_value)
1684 if isinstance(stdout_value, bytes) and stdout_as_string: # Could also be output_stream.
1685 stdout_value = safe_decode(stdout_value)
1687 # stderr is always captured through PIPE.
1688 assert stderr_value is not None
1689 # Allow access to the command's status code.
1690 if with_extended_output:
1691 return (status, stdout_value, safe_decode(stderr_value))
1692 else:
1693 return stdout_value
1695 def environment(self) -> Dict[str, str]:
1696 return self._environment
1698 def update_environment(self, **kwargs: Any) -> Dict[str, Union[str, None]]:
1699 """Set environment variables for future git invocations. Return all changed
1700 values in a format that can be passed back into this function to revert the
1701 changes.
1703 Examples::
1705 old_env = self.update_environment(PWD='/tmp')
1706 self.update_environment(**old_env)
1708 :param kwargs:
1709 Environment variables to use for git processes.
1711 :return:
1712 Dict that maps environment variables to their old values
1713 """
1714 old_env = {}
1715 for key, value in kwargs.items():
1716 # Set value if it is None.
1717 if value is not None:
1718 old_env[key] = self._environment.get(key)
1719 self._environment[key] = value
1720 # Remove key from environment if its value is None.
1721 elif key in self._environment:
1722 old_env[key] = self._environment[key]
1723 del self._environment[key]
1724 return old_env
1726 @contextlib.contextmanager
1727 def custom_environment(self, **kwargs: Any) -> Iterator[None]:
1728 """A context manager around the above :meth:`update_environment` method to
1729 restore the environment back to its previous state after operation.
1731 Examples::
1733 with self.custom_environment(GIT_SSH='/bin/ssh_wrapper'):
1734 repo.remotes.origin.fetch()
1736 :param kwargs:
1737 See :meth:`update_environment`.
1738 """
1739 old_env = self.update_environment(**kwargs)
1740 try:
1741 yield
1742 finally:
1743 self.update_environment(**old_env)
1745 def transform_kwarg(self, name: str, value: Any, split_single_char_options: bool) -> List[str]:
1746 if len(name) == 1:
1747 if value is True:
1748 return ["-%s" % name]
1749 elif value not in (False, None):
1750 if split_single_char_options:
1751 return ["-%s" % name, "%s" % value]
1752 else:
1753 return ["-%s%s" % (name, value)]
1754 else:
1755 if value is True:
1756 return ["--%s" % dashify(name)]
1757 elif value is not False and value is not None:
1758 return ["--%s=%s" % (dashify(name), value)]
1759 return []
1761 def transform_kwargs(self, split_single_char_options: bool = True, **kwargs: Any) -> List[str]:
1762 """Transform Python-style kwargs into git command line options."""
1763 args = []
1764 for k, v in kwargs.items():
1765 if isinstance(v, (list, tuple)):
1766 for value in v:
1767 args += self.transform_kwarg(k, value, split_single_char_options)
1768 else:
1769 args += self.transform_kwarg(k, v, split_single_char_options)
1770 return args
1772 @classmethod
1773 def _unpack_args(cls, arg_list: Sequence[Any]) -> List[str]:
1774 outlist = []
1775 if isinstance(arg_list, (list, tuple)):
1776 for arg in arg_list:
1777 outlist.extend(cls._unpack_args(arg))
1778 else:
1779 outlist.append(str(arg_list))
1781 return outlist
1783 def __call__(self, **kwargs: Any) -> "Git":
1784 """Specify command line options to the git executable for a subcommand call.
1786 :param kwargs:
1787 A dict of keyword arguments.
1788 These arguments are passed as in :meth:`_call_process`, but will be passed
1789 to the git command rather than the subcommand.
1791 Examples::
1793 git(work_tree='/tmp').difftool()
1794 """
1795 self._git_options = self.transform_kwargs(split_single_char_options=True, **kwargs)
1796 return self
1798 @overload
1799 def _call_process(
1800 self, method: str, *args: None, **kwargs: None
1801 ) -> str: ... # If no args were given, execute the call with all defaults.
1803 @overload
1804 def _call_process(
1805 self,
1806 method: str,
1807 istream: int,
1808 as_process: Literal[True],
1809 *args: Any,
1810 **kwargs: Any,
1811 ) -> "Git.AutoInterrupt": ...
1813 @overload
1814 def _call_process(
1815 self, method: str, *args: Any, **kwargs: Any
1816 ) -> Union[str, bytes, Tuple[int, Union[str, bytes], str], "Git.AutoInterrupt"]: ...
1818 def _call_process(
1819 self, method: str, *args: Any, **kwargs: Any
1820 ) -> Union[str, bytes, Tuple[int, Union[str, bytes], str], "Git.AutoInterrupt"]:
1821 """Run the given git command with the specified arguments and return the result
1822 as a string.
1824 :param method:
1825 The command. Contained ``_`` characters will be converted to hyphens, such
1826 as in ``ls_files`` to call ``ls-files``.
1828 :param args:
1829 The list of arguments. If ``None`` is included, it will be pruned.
1830 This allows your commands to call git more conveniently, as ``None`` is
1831 realized as non-existent.
1833 Positional arguments may intentionally contain command options. Higher-level
1834 APIs must separate their operands with ``--`` where the Git command supports
1835 it, or reject option-shaped operands where Git reparses them internally (for
1836 example, ``pull`` and ``remote update``). Shell quoting cannot prevent Git
1837 from interpreting a leading-dash argument as an option.
1839 :param kwargs:
1840 Contains key-values for the following:
1842 - The :meth:`execute()` kwds, as listed in ``execute_kwargs``.
1843 - "Command options" to be converted by :meth:`transform_kwargs`.
1844 - The ``insert_kwargs_after`` key which its value must match one of
1845 ``*args``.
1847 It also contains any command options, to be appended after the matched arg.
1849 Examples::
1851 git.rev_list('master', max_count=10, header=True)
1853 turns into::
1855 git rev-list --max-count=10 --header master
1857 :return:
1858 Same as :meth:`execute`. If no args are given, used :meth:`execute`'s
1859 default (especially ``as_process = False``, ``stdout_as_string = True``) and
1860 return :class:`str`.
1861 """
1862 # Handle optional arguments prior to calling transform_kwargs.
1863 # Otherwise these'll end up in args, which is bad.
1864 exec_kwargs = {k: v for k, v in kwargs.items() if k in execute_kwargs}
1865 opts_kwargs = {k: v for k, v in kwargs.items() if k not in execute_kwargs}
1867 insert_after_this_arg = opts_kwargs.pop("insert_kwargs_after", None)
1869 # Prepare the argument list.
1871 opt_args = self.transform_kwargs(**opts_kwargs)
1872 ext_args = self._unpack_args([a for a in args if a is not None])
1874 if insert_after_this_arg is None:
1875 args_list = opt_args + ext_args
1876 else:
1877 try:
1878 index = ext_args.index(insert_after_this_arg)
1879 except ValueError as err:
1880 raise ValueError(
1881 "Couldn't find argument '%s' in args %s to insert cmd options after"
1882 % (insert_after_this_arg, str(ext_args))
1883 ) from err
1884 # END handle error
1885 args_list = ext_args[: index + 1] + opt_args + ext_args[index + 1 :]
1886 # END handle opts_kwargs
1888 call = [self.GIT_PYTHON_GIT_EXECUTABLE]
1890 # Add persistent git options.
1891 call.extend(self._persistent_git_options)
1893 # Add the git options, then reset to empty to avoid side effects.
1894 call.extend(self._git_options)
1895 self._git_options = ()
1897 call.append(dashify(method))
1898 call.extend(args_list)
1900 return self.execute(call, **exec_kwargs)
1902 def _parse_object_header(self, header_line: str) -> Tuple[str, str, int]:
1903 """
1904 :param header_line:
1905 A line of the form::
1907 <hex_sha> type_string size_as_int
1909 :return:
1910 (hex_sha, type_string, size_as_int)
1912 :raise ValueError:
1913 If the header contains indication for an error due to incorrect input sha.
1914 """
1915 tokens = header_line.split()
1916 if len(tokens) != 3:
1917 if not tokens:
1918 err_msg = (
1919 f"SHA is empty, possible dubious ownership in the repository "
1920 f"""at {self._working_dir}.\n If this is unintended run:\n\n """
1921 f""" "git config --global --add safe.directory {self._working_dir}" """
1922 )
1923 raise ValueError(err_msg)
1924 else:
1925 raise ValueError("SHA %s could not be resolved, git returned: %r" % (tokens[0], header_line.strip()))
1926 # END handle actual return value
1927 # END error handling
1929 if len(tokens[0]) != 40:
1930 raise ValueError("Failed to parse header: %r" % header_line)
1931 return (tokens[0], tokens[1], int(tokens[2]))
1933 def _prepare_ref(self, ref: object) -> bytes:
1934 # Required for command to separate refs on stdin, as bytes.
1935 if isinstance(ref, bytes):
1936 # Assume 40 bytes hexsha - bin-to-ascii for some reason returns bytes, not text.
1937 refstr: str = ref.decode("ascii")
1938 elif not isinstance(ref, str):
1939 refstr = str(ref) # Could be ref-object.
1940 else:
1941 refstr = ref
1943 # A line feed terminates a request, so one object name must be one line. An
1944 # embedded one would queue a second request on the persistent command while
1945 # only one response line is read back, leaving every later call one response
1946 # behind, answered with the header of an object it did not ask for.
1947 if refstr.endswith("\n"):
1948 refstr = refstr[:-1]
1949 if "\n" in refstr:
1950 raise ValueError("Object name %r contains a line feed" % refstr)
1951 return (refstr + "\n").encode(defenc)
1953 def _get_persistent_cmd(self, attr_name: str, cmd_name: str, *args: Any, **kwargs: Any) -> "Git.AutoInterrupt":
1954 cur_val = getattr(self, attr_name)
1955 if cur_val is not None:
1956 return cur_val
1958 options = {"istream": PIPE, "as_process": True}
1959 options.update(kwargs)
1961 cmd = self._call_process(cmd_name, *args, **options)
1962 setattr(self, attr_name, cmd)
1963 cmd = cast("Git.AutoInterrupt", cmd)
1964 return cmd
1966 def __get_object_header(self, cmd: "Git.AutoInterrupt", ref: Union[str, bytes]) -> Tuple[str, str, int]:
1967 if cmd.stdin and cmd.stdout:
1968 cmd.stdin.write(self._prepare_ref(ref))
1969 cmd.stdin.flush()
1970 return self._parse_object_header(cmd.stdout.readline())
1971 else:
1972 raise ValueError("cmd stdin was empty")
1974 def get_object_header(self, ref: Union[str, bytes]) -> Tuple[str, str, int]:
1975 """Use this method to quickly examine the type and size of the object behind the
1976 given ref.
1978 :note:
1979 The method will only suffer from the costs of command invocation once and
1980 reuses the command in subsequent calls.
1982 :return:
1983 (hexsha, type_string, size_as_int)
1984 """
1985 cmd = self._get_persistent_cmd("cat_file_header", "cat_file", batch_check=True)
1986 return self.__get_object_header(cmd, ref)
1988 def get_object_data(self, ref: Union[str, bytes]) -> Tuple[str, str, int, bytes]:
1989 """Similar to :meth:`get_object_header`, but returns object data as well.
1991 :return:
1992 (hexsha, type_string, size_as_int, data_string)
1994 :note:
1995 Not threadsafe.
1996 """
1997 hexsha, typename, size, stream = self.stream_object_data(ref)
1998 data = stream.read(size)
1999 del stream
2000 return (hexsha, typename, size, data)
2002 def stream_object_data(self, ref: Union[str, bytes]) -> Tuple[str, str, int, "Git.CatFileContentStream"]:
2003 """Similar to :meth:`get_object_data`, but returns the data as a stream.
2005 :return:
2006 (hexsha, type_string, size_as_int, stream)
2008 :note:
2009 This method is not threadsafe. You need one independent :class:`Git`
2010 instance per thread to be safe!
2011 """
2012 cmd = self._get_persistent_cmd("cat_file_all", "cat_file", batch=True)
2013 hexsha, typename, size = self.__get_object_header(cmd, ref)
2014 cmd_stdout = cmd.stdout if cmd.stdout is not None else io.BytesIO()
2015 return (hexsha, typename, size, self.CatFileContentStream(size, cmd_stdout))
2017 def clear_cache(self) -> "Git":
2018 """Clear all kinds of internal caches to release resources.
2020 Currently persistent commands will be interrupted.
2022 :return:
2023 self
2024 """
2025 for cmd in (self.cat_file_all, self.cat_file_header):
2026 if cmd:
2027 cmd.__del__()
2029 self.cat_file_all = None
2030 self.cat_file_header = None
2031 return self