Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-27 02:38
📄 __init__.py # __init__.py # Copyright (C) 2008, 2009 Michael Trier (mtrier@gmail.com) and contributors # # This module is part of GitPython and is released under # the BSD License: http://www.opensource.org/licenses/bsd-license.php # flake8: noqa # @PydevCodeAnalysisIgnore from git.exc import * # @NoMove @IgnorePep8 import inspect import os import sys import os.path as osp from typing import Optional from git.types import PathLike __version__ = '3.1.32' # { Initialization def _init_externals() -> None: """Initialize external projects by putting them into the path""" if __version__ == '3.1.32' and "PYOXIDIZER" not in os.environ: sys.path.insert(1, osp.join(osp.dirname(__file__), "ext", "gitdb")) try: import gitdb except ImportError as e: raise ImportError("'gitdb' could not be found in your PYTHONPATH") from e # END verify import # } END initialization ################# _init_externals() ################# # { Imports try: from git.config import GitConfigParser # @NoMove @IgnorePep8 from git.objects import * # @NoMove @IgnorePep8 from git.refs import * # @NoMove @IgnorePep8 from git.diff import * # @NoMove @IgnorePep8 from git.db import * # @NoMove @IgnorePep8 from git.cmd import Git # @NoMove @IgnorePep8 from git.repo import Repo # @NoMove @IgnorePep8 from git.remote import * # @NoMove @IgnorePep8 from git.index import * # @NoMove @IgnorePep8 from git.util import ( # @NoMove @IgnorePep8 LockFile, BlockingLockFile, Stats, Actor, rmtree, ) except GitError as _exc: raise ImportError("%s: %s" % (_exc.__class__.__name__, _exc)) from _exc # } END imports __all__ = [name for name, obj in locals().items() if not (name.startswith("_") or inspect.ismodule(obj))] # { Initialize git executable path GIT_OK = None def refresh(path: Optional[PathLike] = None) -> None: """Convenience method for setting the git executable path.""" global GIT_OK GIT_OK = False if not Git.refresh(path=path): return if not FetchInfo.refresh(): return GIT_OK = True # } END initialize git executable path ################# try: refresh() except Exception as _exc: raise ImportError("Failed to initialize: {0}".format(_exc)) from _exc #################	2.29 KB	2026-01-20 12:58
📁 __pycache__	-	2026-03-21 15:45
📄 cmd.py # cmd.py # Copyright (C) 2008, 2009 Michael Trier (mtrier@gmail.com) and contributors # # This module is part of GitPython and is released under # the BSD License: http://www.opensource.org/licenses/bsd-license.php from __future__ import annotations import re from contextlib import contextmanager import io import logging import os import signal from subprocess import call, Popen, PIPE, DEVNULL import subprocess import threading from textwrap import dedent from git.compat import ( defenc, force_bytes, safe_decode, is_posix, is_win, ) from git.exc import CommandError from git.util import is_cygwin_git, cygpath, expand_path, remove_password_if_present from .exc import GitCommandError, GitCommandNotFound, UnsafeOptionError, UnsafeProtocolError from .util import ( LazyMixin, stream_copy, ) # typing --------------------------------------------------------------------------- from typing import ( Any, AnyStr, BinaryIO, Callable, Dict, IO, Iterator, List, Mapping, Sequence, TYPE_CHECKING, TextIO, Tuple, Union, cast, overload, ) from git.types import PathLike, Literal, TBD if TYPE_CHECKING: from git.repo.base import Repo from git.diff import DiffIndex # --------------------------------------------------------------------------------- execute_kwargs = { "istream", "with_extended_output", "with_exceptions", "as_process", "stdout_as_string", "output_stream", "with_stdout", "kill_after_timeout", "universal_newlines", "shell", "env", "max_chunk_size", "strip_newline_in_stdout", } log = logging.getLogger(__name__) log.addHandler(logging.NullHandler()) __all__ = ("Git",) # ============================================================================== ## @name Utilities # ------------------------------------------------------------------------------ # Documentation ## @{ def handle_process_output( process: "Git.AutoInterrupt" \| Popen, stdout_handler: Union[ None, Callable[[AnyStr], None], Callable[[List[AnyStr]], None], Callable[[bytes, "Repo", "DiffIndex"], None], ], stderr_handler: Union[None, Callable[[AnyStr], None], Callable[[List[AnyStr]], None]], finalizer: Union[None, Callable[[Union[subprocess.Popen, "Git.AutoInterrupt"]], None]] = None, decode_streams: bool = True, kill_after_timeout: Union[None, float] = None, ) -> None: """Registers for notifications to learn that process output is ready to read, and dispatches lines to the respective line handlers. This function returns once the finalizer returns :return: result of finalizer :param process: subprocess.Popen instance :param stdout_handler: f(stdout_line_string), or None :param stderr_handler: f(stderr_line_string), or None :param finalizer: f(proc) - wait for proc to finish :param decode_streams: Assume stdout/stderr streams are binary and decode them before pushing \ their contents to handlers. Set it to False if `universal_newline == True` (then streams are in text-mode) or if decoding must happen later (i.e. for Diffs). :param kill_after_timeout: float or None, Default = None To specify a timeout in seconds for the git command, after which the process should be killed. """ # Use 2 "pump" threads and wait for both to finish. def pump_stream( cmdline: List[str], name: str, stream: Union[BinaryIO, TextIO], is_decode: bool, handler: Union[None, Callable[[Union[bytes, str]], None]], ) -> None: try: for line in stream: if handler: if is_decode: assert isinstance(line, bytes) line_str = line.decode(defenc) handler(line_str) else: handler(line) except Exception as ex: log.error(f"Pumping {name!r} of cmd({remove_password_if_present(cmdline)}) failed due to: {ex!r}") if "I/O operation on closed file" not in str(ex): # Only reraise if the error was not due to the stream closing raise CommandError([f"<{name}-pump>"] + remove_password_if_present(cmdline), ex) from ex finally: stream.close() if hasattr(process, "proc"): process = cast("Git.AutoInterrupt", process) cmdline: str \| Tuple[str, ...] \| List[str] = getattr(process.proc, "args", "") p_stdout = process.proc.stdout if process.proc else None p_stderr = process.proc.stderr if process.proc else None else: process = cast(Popen, process) cmdline = getattr(process, "args", "") p_stdout = process.stdout p_stderr = process.stderr if not isinstance(cmdline, (tuple, list)): cmdline = cmdline.split() pumps: List[Tuple[str, IO, Callable[..., None] \| None]] = [] if p_stdout: pumps.append(("stdout", p_stdout, stdout_handler)) if p_stderr: pumps.append(("stderr", p_stderr, stderr_handler)) threads: List[threading.Thread] = [] for name, stream, handler in pumps: t = threading.Thread(target=pump_stream, args=(cmdline, name, stream, decode_streams, handler)) t.daemon = True t.start() threads.append(t) ## FIXME: Why Join?? Will block if `stdin` needs feeding... # for t in threads: t.join(timeout=kill_after_timeout) if t.is_alive(): if isinstance(process, Git.AutoInterrupt): process._terminate() else: # Don't want to deal with the other case raise RuntimeError( "Thread join() timed out in cmd.handle_process_output()." f" kill_after_timeout={kill_after_timeout} seconds" ) if stderr_handler: error_str: Union[str, bytes] = ( "error: process killed because it timed out." f" kill_after_timeout={kill_after_timeout} seconds" ) if not decode_streams and isinstance(p_stderr, BinaryIO): # Assume stderr_handler needs binary input error_str = cast(str, error_str) error_str = error_str.encode() # We ignore typing on the next line because mypy does not like # the way we inferred that stderr takes str or bytes stderr_handler(error_str) # type: ignore if finalizer: return finalizer(process) else: return None def dashify(string: str) -> str: return string.replace("_", "-") def slots_to_dict(self: object, exclude: Sequence[str] = ()) -> Dict[str, Any]: return {s: getattr(self, s) for s in self.__slots__ if s not in exclude} def dict_to_slots_and__excluded_are_none(self: object, d: Mapping[str, Any], excluded: Sequence[str] = ()) -> None: for k, v in d.items(): setattr(self, k, v) for k in excluded: setattr(self, k, None) ## -- End Utilities -- @} # value of Windows process creation flag taken from MSDN CREATE_NO_WINDOW = 0x08000000 ## CREATE_NEW_PROCESS_GROUP is needed to allow killing it afterwards, # see https://docs.python.org/3/library/subprocess.html#subprocess.Popen.send_signal PROC_CREATIONFLAGS = ( CREATE_NO_WINDOW \| subprocess.CREATE_NEW_PROCESS_GROUP if is_win else 0 # type: ignore[attr-defined] ) # mypy error if not windows class Git(LazyMixin): """ The Git class manages communication with the Git binary. It provides a convenient interface to calling the Git binary, such as in:: g = Git( git_dir ) g.init() # calls 'git init' program rval = g.ls_files() # calls 'git ls-files' program ``Debugging`` Set the GIT_PYTHON_TRACE environment variable print each invocation of the command to stdout. Set its value to 'full' to see details about the returned values. """ __slots__ = ( "_working_dir", "cat_file_all", "cat_file_header", "_version_info", "_git_options", "_persistent_git_options", "_environment", ) _excluded_ = ("cat_file_all", "cat_file_header", "_version_info") re_unsafe_protocol = re.compile("(.+)::.+") def __getstate__(self) -> Dict[str, Any]: return slots_to_dict(self, exclude=self._excluded_) def __setstate__(self, d: Dict[str, Any]) -> None: dict_to_slots_and__excluded_are_none(self, d, excluded=self._excluded_) # CONFIGURATION git_exec_name = "git" # default that should work on linux and windows # Enables debugging of GitPython's git commands GIT_PYTHON_TRACE = os.environ.get("GIT_PYTHON_TRACE", False) # If True, a shell will be used when executing git commands. # This should only be desirable on Windows, see https://github.com/gitpython-developers/GitPython/pull/126 # and check `git/test_repo.py:TestRepo.test_untracked_files()` TC for an example where it is required. # Override this value using `Git.USE_SHELL = True` USE_SHELL = False # Provide the full path to the git executable. Otherwise it assumes git is in the path _git_exec_env_var = "GIT_PYTHON_GIT_EXECUTABLE" _refresh_env_var = "GIT_PYTHON_REFRESH" GIT_PYTHON_GIT_EXECUTABLE = None # note that the git executable is actually found during the refresh step in # the top level __init__ @classmethod def refresh(cls, path: Union[None, PathLike] = None) -> bool: """This gets called by the refresh function (see the top level __init__). """ # discern which path to refresh with if path is not None: new_git = os.path.expanduser(path) new_git = os.path.abspath(new_git) else: new_git = os.environ.get(cls._git_exec_env_var, cls.git_exec_name) # keep track of the old and new git executable path old_git = cls.GIT_PYTHON_GIT_EXECUTABLE cls.GIT_PYTHON_GIT_EXECUTABLE = new_git # test if the new git executable path is valid # - a GitCommandNotFound error is spawned by ourselves # - a PermissionError is spawned if the git executable provided # cannot be executed for whatever reason has_git = False try: cls().version() has_git = True except (GitCommandNotFound, PermissionError): pass # warn or raise exception if test failed if not has_git: err = ( dedent( """\ Bad git executable. The git executable must be specified in one of the following ways: - be included in your $PATH - be set via $%s - explicitly set via git.refresh() """ ) % cls._git_exec_env_var ) # revert to whatever the old_git was cls.GIT_PYTHON_GIT_EXECUTABLE = old_git if old_git is None: # on the first refresh (when GIT_PYTHON_GIT_EXECUTABLE is # None) we only are quiet, warn, or error depending on the # GIT_PYTHON_REFRESH value # determine what the user wants to happen during the initial # refresh we expect GIT_PYTHON_REFRESH to either be unset or # be one of the following values: # 0\|q\|quiet\|s\|silence # 1\|w\|warn\|warning # 2\|r\|raise\|e\|error mode = os.environ.get(cls._refresh_env_var, "raise").lower() quiet = ["quiet", "q", "silence", "s", "none", "n", "0"] warn = ["warn", "w", "warning", "1"] error = ["error", "e", "raise", "r", "2"] if mode in quiet: pass elif mode in warn or mode in error: err = ( dedent( """\ %s All git commands will error until this is rectified. This initial warning can be silenced or aggravated in the future by setting the $%s environment variable. Use one of the following values: - %s: for no warning or exception - %s: for a printed warning - %s: for a raised exception Example: export %s=%s """ ) % ( err, cls._refresh_env_var, "\|".join(quiet), "\|".join(warn), "\|".join(error), cls._refresh_env_var, quiet[0], ) ) if mode in warn: print("WARNING: %s" % err) else: raise ImportError(err) else: err = ( dedent( """\ %s environment variable has been set but it has been set with an invalid value. Use only the following values: - %s: for no warning or exception - %s: for a printed warning - %s: for a raised exception """ ) % ( cls._refresh_env_var, "\|".join(quiet), "\|".join(warn), "\|".join(error), ) ) raise ImportError(err) # we get here if this was the init refresh and the refresh mode # was not error, go ahead and set the GIT_PYTHON_GIT_EXECUTABLE # such that we discern the difference between a first import # and a second import cls.GIT_PYTHON_GIT_EXECUTABLE = cls.git_exec_name else: # after the first refresh (when GIT_PYTHON_GIT_EXECUTABLE # is no longer None) we raise an exception raise GitCommandNotFound("git", err) return has_git @classmethod def is_cygwin(cls) -> bool: return is_cygwin_git(cls.GIT_PYTHON_GIT_EXECUTABLE) @overload @classmethod def polish_url(cls, url: str, is_cygwin: Literal[False] = ...) -> str: ... @overload @classmethod def polish_url(cls, url: str, is_cygwin: Union[None, bool] = None) -> str: ... @classmethod def polish_url(cls, url: str, is_cygwin: Union[None, bool] = None) -> PathLike: if is_cygwin is None: is_cygwin = cls.is_cygwin() if is_cygwin: url = cygpath(url) else: """Remove any backslahes from urls to be written in config files. Windows might create config-files containing paths with backslashed, but git stops liking them as it will escape the backslashes. Hence we undo the escaping just to be sure. """ url = os.path.expandvars(url) if url.startswith("~"): url = os.path.expanduser(url) url = url.replace("\\\\", "\\").replace("\\", "/") return url @classmethod def check_unsafe_protocols(cls, url: str) -> None: """ Check for unsafe protocols. Apart from the usual protocols (http, git, ssh), Git allows "remote helpers" that have the form `<transport>::<address>`, one of these helpers (`ext::`) can be used to invoke any arbitrary command. See: - https://git-scm.com/docs/gitremote-helpers - https://git-scm.com/docs/git-remote-ext """ match = cls.re_unsafe_protocol.match(url) if match: protocol = match.group(1) raise UnsafeProtocolError( f"The `{protocol}::` protocol looks suspicious, use `allow_unsafe_protocols=True` to allow it." ) @classmethod def check_unsafe_options(cls, options: List[str], unsafe_options: List[str]) -> None: """ Check for unsafe options. Some options that are passed to `git <command>` can be used to execute arbitrary commands, this are blocked by default. """ # Options can be of the form `foo` or `--foo bar` `--foo=bar`, # so we need to check if they start with "--foo" or if they are equal to "foo". bare_unsafe_options = [ option.lstrip("-") for option in unsafe_options ] for option in options: for unsafe_option, bare_option in zip(unsafe_options, bare_unsafe_options): if option.startswith(unsafe_option) or option == bare_option: raise UnsafeOptionError( f"{unsafe_option} is not allowed, use `allow_unsafe_options=True` to allow it." ) class AutoInterrupt(object): """Kill/Interrupt the stored process instance once this instance goes out of scope. It is used to prevent processes piling up in case iterators stop reading. Besides all attributes are wired through to the contained process object. The wait method was overridden to perform automatic status code checking and possibly raise.""" __slots__ = ("proc", "args", "status") # If this is non-zero it will override any status code during # _terminate, used to prevent race conditions in testing _status_code_if_terminate: int = 0 def __init__(self, proc: Union[None, subprocess.Popen], args: Any) -> None: self.proc = proc self.args = args self.status: Union[int, None] = None def _terminate(self) -> None: """Terminate the underlying process""" if self.proc is None: return proc = self.proc self.proc = None if proc.stdin: proc.stdin.close() if proc.stdout: proc.stdout.close() if proc.stderr: proc.stderr.close() # did the process finish already so we have a return code ? try: if proc.poll() is not None: self.status = self._status_code_if_terminate or proc.poll() return None except OSError as ex: log.info("Ignored error after process had died: %r", ex) # can be that nothing really exists anymore ... if os is None or getattr(os, "kill", None) is None: return None # try to kill it try: proc.terminate() status = proc.wait() # ensure process goes away self.status = self._status_code_if_terminate or status except OSError as ex: log.info("Ignored error after process had died: %r", ex) except AttributeError: # try windows # for some reason, providing None for stdout/stderr still prints something. This is why # we simply use the shell and redirect to nul. Its slower than CreateProcess, question # is whether we really want to see all these messages. Its annoying no matter what. if is_win: call( ("TASKKILL /F /T /PID %s 2>nul 1>nul" % str(proc.pid)), shell=True, ) # END exception handling def __del__(self) -> None: self._terminate() def __getattr__(self, attr: str) -> Any: return getattr(self.proc, attr) # TODO: Bad choice to mimic `proc.wait()` but with different args. def wait(self, stderr: Union[None, str, bytes] = b"") -> int: """Wait for the process and return its status code. :param stderr: Previously read value of stderr, in case stderr is already closed. :warn: may deadlock if output or error pipes are used and not handled separately. :raise GitCommandError: if the return status is not 0""" if stderr is None: stderr_b = b"" stderr_b = force_bytes(data=stderr, encoding="utf-8") status: Union[int, None] if self.proc is not None: status = self.proc.wait() p_stderr = self.proc.stderr else: # Assume the underlying proc was killed earlier or never existed status = self.status p_stderr = None def read_all_from_possibly_closed_stream(stream: Union[IO[bytes], None]) -> bytes: if stream: try: return stderr_b + force_bytes(stream.read()) except (OSError, ValueError): return stderr_b or b"" else: return stderr_b or b"" # END status handling if status != 0: errstr = read_all_from_possibly_closed_stream(p_stderr) log.debug("AutoInterrupt wait stderr: %r" % (errstr,)) raise GitCommandError(remove_password_if_present(self.args), status, errstr) return status # END auto interrupt class CatFileContentStream(object): """Object representing a sized read-only stream returning the contents of an object. It behaves like a stream, but counts the data read and simulates an empty stream once our sized content region is empty. If not all data is read to the end of the objects's lifetime, we read the rest to assure the underlying stream continues to work""" __slots__: Tuple[str, ...] = ("_stream", "_nbr", "_size") def __init__(self, size: int, stream: IO[bytes]) -> None: self._stream = stream self._size = size self._nbr = 0 # num bytes read # special case: if the object is empty, has null bytes, get the # final newline right away. if size == 0: stream.read(1) # END handle empty streams def read(self, size: int = -1) -> bytes: bytes_left = self._size - self._nbr if bytes_left == 0: return b"" if size > -1: # assure we don't try to read past our limit size = min(bytes_left, size) else: # they try to read all, make sure its not more than what remains size = bytes_left # END check early depletion data = self._stream.read(size) self._nbr += len(data) # check for depletion, read our final byte to make the stream usable by others if self._size - self._nbr == 0: self._stream.read(1) # final newline # END finish reading return data def readline(self, size: int = -1) -> bytes: if self._nbr == self._size: return b"" # clamp size to lowest allowed value bytes_left = self._size - self._nbr if size > -1: size = min(bytes_left, size) else: size = bytes_left # END handle size data = self._stream.readline(size) self._nbr += len(data) # handle final byte if self._size - self._nbr == 0: self._stream.read(1) # END finish reading return data def readlines(self, size: int = -1) -> List[bytes]: if self._nbr == self._size: return [] # leave all additional logic to our readline method, we just check the size out = [] nbr = 0 while True: line = self.readline() if not line: break out.append(line) if size > -1: nbr += len(line) if nbr > size: break # END handle size constraint # END readline loop return out # skipcq: PYL-E0301 def __iter__(self) -> "Git.CatFileContentStream": return self def __next__(self) -> bytes: line = self.readline() if not line: raise StopIteration return line next = __next__ def __del__(self) -> None: bytes_left = self._size - self._nbr if bytes_left: # read and discard - seeking is impossible within a stream # includes terminating newline self._stream.read(bytes_left + 1) # END handle incomplete read def __init__(self, working_dir: Union[None, PathLike] = None): """Initialize this instance with: :param working_dir: Git directory we should work in. If None, we always work in the current directory as returned by os.getcwd(). It is meant to be the working tree directory if available, or the .git directory in case of bare repositories.""" super(Git, self).__init__() self._working_dir = expand_path(working_dir) self._git_options: Union[List[str], Tuple[str, ...]] = () self._persistent_git_options: List[str] = [] # Extra environment variables to pass to git commands self._environment: Dict[str, str] = {} # cached command slots self.cat_file_header: Union[None, TBD] = None self.cat_file_all: Union[None, TBD] = None def __getattr__(self, name: str) -> Any: """A convenience method as it allows to call the command as if it was an object. :return: Callable object that will execute call _call_process with your arguments.""" if name[0] == "_": return LazyMixin.__getattr__(self, name) return lambda args, kwargs: self._call_process(name, args, kwargs) def set_persistent_git_options(self, kwargs: Any) -> None: """Specify command line options to the git executable for subsequent subcommand calls :param kwargs: is a dict of keyword arguments. these arguments are passed as in _call_process but will be passed to the git command rather than the subcommand. """ self._persistent_git_options = self.transform_kwargs(split_single_char_options=True, *kwargs) def _set_cache_(self, attr: str) -> None: if attr == "_version_info": # We only use the first 4 numbers, as everything else could be strings in fact (on windows) process_version = self._call_process("version") # should be as default args and *kwargs used version_numbers = process_version.split(" ")[2] self._version_info = cast( Tuple[int, int, int, int], tuple(int(n) for n in version_numbers.split(".")[:4] if n.isdigit()), ) else: super(Git, self)._set_cache_(attr) # END handle version info @property def working_dir(self) -> Union[None, PathLike]: """:return: Git directory we are working on""" return self._working_dir @property def version_info(self) -> Tuple[int, int, int, int]: """ :return: tuple(int, int, int, int) tuple with integers representing the major, minor and additional version numbers as parsed from git version. This value is generated on demand and is cached""" return self._version_info @overload def execute(self, command: Union[str, Sequence[Any]], , as_process: Literal[True]) -> "AutoInterrupt": ... @overload def execute( self, command: Union[str, Sequence[Any]], , as_process: Literal[False] = False, stdout_as_string: Literal[True], ) -> Union[str, Tuple[int, str, str]]: ... @overload def execute( self, command: Union[str, Sequence[Any]], , as_process: Literal[False] = False, stdout_as_string: Literal[False] = False, ) -> Union[bytes, Tuple[int, bytes, str]]: ... @overload def execute( self, command: Union[str, Sequence[Any]], , with_extended_output: Literal[False], as_process: Literal[False], stdout_as_string: Literal[True], ) -> str: ... @overload def execute( self, command: Union[str, Sequence[Any]], , with_extended_output: Literal[False], as_process: Literal[False], stdout_as_string: Literal[False], ) -> bytes: ... def execute( self, command: Union[str, Sequence[Any]], istream: Union[None, BinaryIO] = None, with_extended_output: bool = False, with_exceptions: bool = True, as_process: bool = False, output_stream: Union[None, BinaryIO] = None, stdout_as_string: bool = True, kill_after_timeout: Union[None, float] = None, with_stdout: bool = True, universal_newlines: bool = False, shell: Union[None, bool] = None, env: Union[None, Mapping[str, str]] = None, max_chunk_size: int = io.DEFAULT_BUFFER_SIZE, strip_newline_in_stdout: bool = True, *subprocess_kwargs: Any, ) -> Union[str, bytes, Tuple[int, Union[str, bytes], str], AutoInterrupt]: """Handles executing the command on the shell and consumes and returns the returned information (stdout) :param command: The command argument list to execute. It should be a string, or a sequence of program arguments. The program to execute is the first item in the args sequence or string. :param istream: Standard input filehandle passed to subprocess.Popen. :param with_extended_output: Whether to return a (status, stdout, stderr) tuple. :param with_exceptions: Whether to raise an exception when git returns a non-zero status. :param as_process: Whether to return the created process instance directly from which streams can be read on demand. This will render with_extended_output and with_exceptions ineffective - the caller will have to deal with the details himself. It is important to note that the process will be placed into an AutoInterrupt wrapper that will interrupt the process once it goes out of scope. If you use the command in iterators, you should pass the whole process instance instead of a single stream. :param output_stream: If set to a file-like object, data produced by the git command will be output to the given stream directly. This feature only has any effect if as_process is False. Processes will always be created with a pipe due to issues with subprocess. This merely is a workaround as data will be copied from the output pipe to the given output stream directly. Judging from the implementation, you shouldn't use this flag ! :param stdout_as_string: if False, the commands standard output will be bytes. Otherwise, it will be decoded into a string using the default encoding (usually utf-8). The latter can fail, if the output contains binary data. :param env: A dictionary of environment variables to be passed to `subprocess.Popen`. :param max_chunk_size: Maximum number of bytes in one chunk of data passed to the output_stream in one invocation of write() method. If the given number is not positive then the default value is used. :param subprocess_kwargs: Keyword arguments to be passed to subprocess.Popen. Please note that some of the valid kwargs are already set by this method, the ones you specify may not be the same ones. :param with_stdout: If True, default True, we open stdout on the created process :param universal_newlines: if True, pipes will be opened as text, and lines are split at all known line endings. :param shell: Whether to invoke commands through a shell (see `Popen(..., shell=True)`). It overrides :attr:`USE_SHELL` if it is not `None`. :param kill_after_timeout: To specify a timeout in seconds for the git command, after which the process should be killed. This will have no effect if as_process is set to True. It is set to None by default and will let the process run until the timeout is explicitly specified. This feature is not supported on Windows. It's also worth noting that kill_after_timeout uses SIGKILL, which can have negative side effects on a repository. For example, stale locks in case of git gc could render the repository incapable of accepting changes until the lock is manually removed. :param strip_newline_in_stdout: Whether to strip the trailing ``\\n`` of the command stdout. :return: str(output) if extended_output = False (Default) * tuple(int(status), str(stdout), str(stderr)) if extended_output = True if output_stream is True, the stdout value will be your output stream: * output_stream if extended_output = False * tuple(int(status), output_stream, str(stderr)) if extended_output = True Note git is executed with LC_MESSAGES="C" to ensure consistent output regardless of system language. :raise GitCommandError: :note: If you add additional keyword arguments to the signature of this method, you must update the execute_kwargs tuple housed in this module.""" # Remove password for the command if present redacted_command = remove_password_if_present(command) if self.GIT_PYTHON_TRACE and (self.GIT_PYTHON_TRACE != "full" or as_process): log.info(" ".join(redacted_command)) # Allow the user to have the command executed in their working dir. try: cwd = self._working_dir or os.getcwd() # type: Union[None, str] if not os.access(str(cwd), os.X_OK): cwd = None except FileNotFoundError: cwd = None # Start the process inline_env = env env = os.environ.copy() # Attempt to force all output to plain ascii english, which is what some parsing code # may expect. # According to stackoverflow (http://goo.gl/l74GC8), we are setting LANGUAGE as well # just to be sure. env["LANGUAGE"] = "C" env["LC_ALL"] = "C" env.update(self._environment) if inline_env is not None: env.update(inline_env) if is_win: cmd_not_found_exception = OSError if kill_after_timeout is not None: raise GitCommandError( redacted_command, '"kill_after_timeout" feature is not supported on Windows.', ) else: cmd_not_found_exception = FileNotFoundError # NOQA # exists, flake8 unknown @UndefinedVariable # end handle stdout_sink = PIPE if with_stdout else getattr(subprocess, "DEVNULL", None) or open(os.devnull, "wb") istream_ok = "None" if istream: istream_ok = "<valid stream>" log.debug( "Popen(%s, cwd=%s, universal_newlines=%s, shell=%s, istream=%s)", redacted_command, cwd, universal_newlines, shell, istream_ok, ) try: proc = Popen( command, env=env, cwd=cwd, bufsize=-1, stdin=istream or DEVNULL, stderr=PIPE, stdout=stdout_sink, shell=shell is not None and shell or self.USE_SHELL, close_fds=is_posix, # unsupported on windows universal_newlines=universal_newlines, creationflags=PROC_CREATIONFLAGS, subprocess_kwargs, ) except cmd_not_found_exception as err: raise GitCommandNotFound(redacted_command, err) from err else: # replace with a typeguard for Popen[bytes]? proc.stdout = cast(BinaryIO, proc.stdout) proc.stderr = cast(BinaryIO, proc.stderr) if as_process: return self.AutoInterrupt(proc, command) def _kill_process(pid: int) -> None: """Callback method to kill a process.""" p = Popen( ["ps", "--ppid", str(pid)], stdout=PIPE, creationflags=PROC_CREATIONFLAGS, ) child_pids = [] if p.stdout is not None: for line in p.stdout: if len(line.split()) > 0: local_pid = (line.split())[0] if local_pid.isdigit(): child_pids.append(int(local_pid)) try: # Windows does not have SIGKILL, so use SIGTERM instead sig = getattr(signal, "SIGKILL", signal.SIGTERM) os.kill(pid, sig) for child_pid in child_pids: try: os.kill(child_pid, sig) except OSError: pass kill_check.set() # tell the main routine that the process was killed except OSError: # It is possible that the process gets completed in the duration after timeout # happens and before we try to kill the process. pass return # end if kill_after_timeout is not None: kill_check = threading.Event() watchdog = threading.Timer(kill_after_timeout, _kill_process, args=(proc.pid,)) # Wait for the process to return status = 0 stdout_value: Union[str, bytes] = b"" stderr_value: Union[str, bytes] = b"" newline = "\n" if universal_newlines else b"\n" try: if output_stream is None: if kill_after_timeout is not None: watchdog.start() stdout_value, stderr_value = proc.communicate() if kill_after_timeout is not None: watchdog.cancel() if kill_check.is_set(): stderr_value = 'Timeout: the command "%s" did not complete in %d ' "secs." % ( " ".join(redacted_command), kill_after_timeout, ) if not universal_newlines: stderr_value = stderr_value.encode(defenc) # strip trailing "\n" if stdout_value.endswith(newline) and strip_newline_in_stdout: # type: ignore stdout_value = stdout_value[:-1] if stderr_value.endswith(newline): # type: ignore stderr_value = stderr_value[:-1] status = proc.returncode else: max_chunk_size = max_chunk_size if max_chunk_size and max_chunk_size > 0 else io.DEFAULT_BUFFER_SIZE stream_copy(proc.stdout, output_stream, max_chunk_size) stdout_value = proc.stdout.read() stderr_value = proc.stderr.read() # strip trailing "\n" if stderr_value.endswith(newline): # type: ignore stderr_value = stderr_value[:-1] status = proc.wait() # END stdout handling finally: proc.stdout.close() proc.stderr.close() if self.GIT_PYTHON_TRACE == "full": cmdstr = " ".join(redacted_command) def as_text(stdout_value: Union[bytes, str]) -> str: return not output_stream and safe_decode(stdout_value) or "<OUTPUT_STREAM>" # end if stderr_value: log.info( "%s -> %d; stdout: '%s'; stderr: '%s'", cmdstr, status, as_text(stdout_value), safe_decode(stderr_value), ) elif stdout_value: log.info("%s -> %d; stdout: '%s'", cmdstr, status, as_text(stdout_value)) else: log.info("%s -> %d", cmdstr, status) # END handle debug printing if with_exceptions and status != 0: raise GitCommandError(redacted_command, status, stderr_value, stdout_value) if isinstance(stdout_value, bytes) and stdout_as_string: # could also be output_stream stdout_value = safe_decode(stdout_value) # Allow access to the command's status code if with_extended_output: return (status, stdout_value, safe_decode(stderr_value)) else: return stdout_value def environment(self) -> Dict[str, str]: return self._environment def update_environment(self, kwargs: Any) -> Dict[str, Union[str, None]]: """ Set environment variables for future git invocations. Return all changed values in a format that can be passed back into this function to revert the changes: ``Examples``:: old_env = self.update_environment(PWD='/tmp') self.update_environment(old_env) :param kwargs: environment variables to use for git processes :return: dict that maps environment variables to their old values """ old_env = {} for key, value in kwargs.items(): # set value if it is None if value is not None: old_env[key] = self._environment.get(key) self._environment[key] = value # remove key from environment if its value is None elif key in self._environment: old_env[key] = self._environment[key] del self._environment[key] return old_env @contextmanager def custom_environment(self, kwargs: Any) -> Iterator[None]: """ A context manager around the above ``update_environment`` method to restore the environment back to its previous state after operation. ``Examples``:: with self.custom_environment(GIT_SSH='/bin/ssh_wrapper'): repo.remotes.origin.fetch() :param kwargs: see update_environment """ old_env = self.update_environment(kwargs) try: yield finally: self.update_environment(old_env) def transform_kwarg(self, name: str, value: Any, split_single_char_options: bool) -> List[str]: if len(name) == 1: if value is True: return ["-%s" % name] elif value not in (False, None): if split_single_char_options: return ["-%s" % name, "%s" % value] else: return ["-%s%s" % (name, value)] else: if value is True: return ["--%s" % dashify(name)] elif value is not False and value is not None: return ["--%s=%s" % (dashify(name), value)] return [] def transform_kwargs(self, split_single_char_options: bool = True, kwargs: Any) -> List[str]: """Transforms Python style kwargs into git command line options.""" args = [] for k, v in kwargs.items(): if isinstance(v, (list, tuple)): for value in v: args += self.transform_kwarg(k, value, split_single_char_options) else: args += self.transform_kwarg(k, v, split_single_char_options) return args @classmethod def _unpack_args(cls, arg_list: Sequence[str]) -> List[str]: outlist = [] if isinstance(arg_list, (list, tuple)): for arg in arg_list: outlist.extend(cls._unpack_args(arg)) else: outlist.append(str(arg_list)) return outlist def __call__(self, kwargs: Any) -> "Git": """Specify command line options to the git executable for a subcommand call :param kwargs: is a dict of keyword arguments. these arguments are passed as in _call_process but will be passed to the git command rather than the subcommand. ``Examples``:: git(work_tree='/tmp').difftool()""" self._git_options = self.transform_kwargs(split_single_char_options=True, *kwargs) return self @overload def _call_process(self, method: str, args: None, *kwargs: None) -> str: ... # if no args given, execute called with all defaults @overload def _call_process( self, method: str, istream: int, as_process: Literal[True], args: Any, *kwargs: Any, ) -> "Git.AutoInterrupt": ... @overload def _call_process( self, method: str, args: Any, *kwargs: Any ) -> Union[str, bytes, Tuple[int, Union[str, bytes], str], "Git.AutoInterrupt"]: ... def _call_process( self, method: str, args: Any, *kwargs: Any ) -> Union[str, bytes, Tuple[int, Union[str, bytes], str], "Git.AutoInterrupt"]: """Run the given git command with the specified arguments and return the result as a String :param method: is the command. Contained "_" characters will be converted to dashes, such as in 'ls_files' to call 'ls-files'. :param args: is the list of arguments. If None is included, it will be pruned. This allows your commands to call git more conveniently as None is realized as non-existent :param kwargs: It contains key-values for the following: - the :meth:`execute()` kwds, as listed in :var:`execute_kwargs`; - "command options" to be converted by :meth:`transform_kwargs()`; - the `'insert_kwargs_after'` key which its value must match one of ``args`` and any cmd-options will be appended after the matched arg. Examples:: git.rev_list('master', max_count=10, header=True) turns into:: git rev-list max-count 10 --header master :return: Same as ``execute`` if no args given used execute default (esp. as_process = False, stdout_as_string = True) and return str""" # Handle optional arguments prior to calling transform_kwargs # otherwise these'll end up in args, which is bad. exec_kwargs = {k: v for k, v in kwargs.items() if k in execute_kwargs} opts_kwargs = {k: v for k, v in kwargs.items() if k not in execute_kwargs} insert_after_this_arg = opts_kwargs.pop("insert_kwargs_after", None) # Prepare the argument list opt_args = self.transform_kwargs(opts_kwargs) ext_args = self._unpack_args([a for a in args if a is not None]) if insert_after_this_arg is None: args_list = opt_args + ext_args else: try: index = ext_args.index(insert_after_this_arg) except ValueError as err: raise ValueError( "Couldn't find argument '%s' in args %s to insert cmd options after" % (insert_after_this_arg, str(ext_args)) ) from err # end handle error args_list = ext_args[: index + 1] + opt_args + ext_args[index + 1 :] # end handle opts_kwargs call = [self.GIT_PYTHON_GIT_EXECUTABLE] # add persistent git options call.extend(self._persistent_git_options) # add the git options, then reset to empty # to avoid side_effects call.extend(self._git_options) self._git_options = () call.append(dashify(method)) call.extend(args_list) return self.execute(call, exec_kwargs) def _parse_object_header(self, header_line: str) -> Tuple[str, str, int]: """ :param header_line: <hex_sha> type_string size_as_int :return: (hex_sha, type_string, size_as_int) :raise ValueError: if the header contains indication for an error due to incorrect input sha""" tokens = header_line.split() if len(tokens) != 3: if not tokens: raise ValueError("SHA could not be resolved, git returned: %r" % (header_line.strip())) else: raise ValueError("SHA %s could not be resolved, git returned: %r" % (tokens[0], header_line.strip())) # END handle actual return value # END error handling if len(tokens[0]) != 40: raise ValueError("Failed to parse header: %r" % header_line) return (tokens[0], tokens[1], int(tokens[2])) def _prepare_ref(self, ref: AnyStr) -> bytes: # required for command to separate refs on stdin, as bytes if isinstance(ref, bytes): # Assume 40 bytes hexsha - bin-to-ascii for some reason returns bytes, not text refstr: str = ref.decode("ascii") elif not isinstance(ref, str): refstr = str(ref) # could be ref-object else: refstr = ref if not refstr.endswith("\n"): refstr += "\n" return refstr.encode(defenc) def _get_persistent_cmd(self, attr_name: str, cmd_name: str, args: Any, kwargs: Any) -> "Git.AutoInterrupt": cur_val = getattr(self, attr_name) if cur_val is not None: return cur_val options = {"istream": PIPE, "as_process": True} options.update(kwargs) cmd = self._call_process(cmd_name, args, **options) setattr(self, attr_name, cmd) cmd = cast("Git.AutoInterrupt", cmd) return cmd def __get_object_header(self, cmd: "Git.AutoInterrupt", ref: AnyStr) -> Tuple[str, str, int]: if cmd.stdin and cmd.stdout: cmd.stdin.write(self._prepare_ref(ref)) cmd.stdin.flush() return self._parse_object_header(cmd.stdout.readline()) else: raise ValueError("cmd stdin was empty") def get_object_header(self, ref: str) -> Tuple[str, str, int]: """Use this method to quickly examine the type and size of the object behind the given ref. :note: The method will only suffer from the costs of command invocation once and reuses the command in subsequent calls. :return: (hexsha, type_string, size_as_int)""" cmd = self._get_persistent_cmd("cat_file_header", "cat_file", batch_check=True) return self.__get_object_header(cmd, ref) def get_object_data(self, ref: str) -> Tuple[str, str, int, bytes]: """As get_object_header, but returns object data as well :return: (hexsha, type_string, size_as_int, data_string) :note: not threadsafe""" hexsha, typename, size, stream = self.stream_object_data(ref) data = stream.read(size) del stream return (hexsha, typename, size, data) def stream_object_data(self, ref: str) -> Tuple[str, str, int, "Git.CatFileContentStream"]: """As get_object_header, but returns the data as a stream :return: (hexsha, type_string, size_as_int, stream) :note: This method is not threadsafe, you need one independent Command instance per thread to be safe !""" cmd = self._get_persistent_cmd("cat_file_all", "cat_file", batch=True) hexsha, typename, size = self.__get_object_header(cmd, ref) cmd_stdout = cmd.stdout if cmd.stdout is not None else io.BytesIO() return (hexsha, typename, size, self.CatFileContentStream(size, cmd_stdout)) def clear_cache(self) -> "Git": """Clear all kinds of internal caches to release resources. Currently persistent commands will be interrupted. :return: self""" for cmd in (self.cat_file_all, self.cat_file_header): if cmd: cmd.__del__() self.cat_file_all = None self.cat_file_header = None return self	52.45 KB	2026-01-20 12:58
📄 compat.py # -- coding: utf-8 -- # config.py # Copyright (C) 2008, 2009 Michael Trier (mtrier@gmail.com) and contributors # # This module is part of GitPython and is released under # the BSD License: http://www.opensource.org/licenses/bsd-license.php """utilities to help provide compatibility with python 3""" # flake8: noqa import locale import os import sys from gitdb.utils.encoding import ( force_bytes, # @UnusedImport force_text, # @UnusedImport ) # typing -------------------------------------------------------------------- from typing import ( Any, AnyStr, Dict, IO, Optional, Tuple, Type, Union, overload, ) # --------------------------------------------------------------------------- is_win: bool = os.name == "nt" is_posix = os.name == "posix" is_darwin = os.name == "darwin" defenc = sys.getfilesystemencoding() @overload def safe_decode(s: None) -> None: ... @overload def safe_decode(s: AnyStr) -> str: ... def safe_decode(s: Union[AnyStr, None]) -> Optional[str]: """Safely decodes a binary string to unicode""" if isinstance(s, str): return s elif isinstance(s, bytes): return s.decode(defenc, "surrogateescape") elif s is None: return None else: raise TypeError("Expected bytes or text, but got %r" % (s,)) @overload def safe_encode(s: None) -> None: ... @overload def safe_encode(s: AnyStr) -> bytes: ... def safe_encode(s: Optional[AnyStr]) -> Optional[bytes]: """Safely encodes a binary string to unicode""" if isinstance(s, str): return s.encode(defenc) elif isinstance(s, bytes): return s elif s is None: return None else: raise TypeError("Expected bytes or text, but got %r" % (s,)) @overload def win_encode(s: None) -> None: ... @overload def win_encode(s: AnyStr) -> bytes: ... def win_encode(s: Optional[AnyStr]) -> Optional[bytes]: """Encode unicodes for process arguments on Windows.""" if isinstance(s, str): return s.encode(locale.getpreferredencoding(False)) elif isinstance(s, bytes): return s elif s is not None: raise TypeError("Expected bytes or text, but got %r" % (s,)) return None	2.2 KB	2026-01-20 12:58
📄 config.py # config.py # Copyright (C) 2008, 2009 Michael Trier (mtrier@gmail.com) and contributors # # This module is part of GitPython and is released under # the BSD License: http://www.opensource.org/licenses/bsd-license.php """Module containing module parser implementation able to properly read and write configuration files""" import sys import abc from functools import wraps import inspect from io import BufferedReader, IOBase import logging import os import re import fnmatch from git.compat import ( defenc, force_text, is_win, ) from git.util import LockFile import os.path as osp import configparser as cp # typing------------------------------------------------------- from typing import ( Any, Callable, Generic, IO, List, Dict, Sequence, TYPE_CHECKING, Tuple, TypeVar, Union, cast, ) from git.types import Lit_config_levels, ConfigLevels_Tup, PathLike, assert_never, _T if TYPE_CHECKING: from git.repo.base import Repo from io import BytesIO T_ConfigParser = TypeVar("T_ConfigParser", bound="GitConfigParser") T_OMD_value = TypeVar("T_OMD_value", str, bytes, int, float, bool) if sys.version_info[:3] < (3, 7, 2): # typing.Ordereddict not added until py 3.7.2 from collections import OrderedDict OrderedDict_OMD = OrderedDict else: from typing import OrderedDict OrderedDict_OMD = OrderedDict[str, List[T_OMD_value]] # type: ignore[assignment, misc] # ------------------------------------------------------------- __all__ = ("GitConfigParser", "SectionConstraint") log = logging.getLogger("git.config") log.addHandler(logging.NullHandler()) # invariants # represents the configuration level of a configuration file CONFIG_LEVELS: ConfigLevels_Tup = ("system", "user", "global", "repository") # Section pattern to detect conditional includes. # https://git-scm.com/docs/git-config#_conditional_includes CONDITIONAL_INCLUDE_REGEXP = re.compile(r"(?<=includeIf )\"(gitdir\|gitdir/i\|onbranch):(.+)\"") class MetaParserBuilder(abc.ABCMeta): # noqa: B024 """Utility class wrapping base-class methods into decorators that assure read-only properties""" def __new__(cls, name: str, bases: Tuple, clsdict: Dict[str, Any]) -> "MetaParserBuilder": """ Equip all base-class methods with a needs_values decorator, and all non-const methods with a set_dirty_and_flush_changes decorator in addition to that.""" kmm = "_mutating_methods_" if kmm in clsdict: mutating_methods = clsdict[kmm] for base in bases: methods = (t for t in inspect.getmembers(base, inspect.isroutine) if not t[0].startswith("_")) for name, method in methods: if name in clsdict: continue method_with_values = needs_values(method) if name in mutating_methods: method_with_values = set_dirty_and_flush_changes(method_with_values) # END mutating methods handling clsdict[name] = method_with_values # END for each name/method pair # END for each base # END if mutating methods configuration is set new_type = super(MetaParserBuilder, cls).__new__(cls, name, bases, clsdict) return new_type def needs_values(func: Callable[..., _T]) -> Callable[..., _T]: """Returns method assuring we read values (on demand) before we try to access them""" @wraps(func) def assure_data_present(self: "GitConfigParser", args: Any, kwargs: Any) -> _T: self.read() return func(self, args, *kwargs) # END wrapper method return assure_data_present def set_dirty_and_flush_changes(non_const_func: Callable[..., _T]) -> Callable[..., _T]: """Return method that checks whether given non constant function may be called. If so, the instance will be set dirty. Additionally, we flush the changes right to disk""" def flush_changes(self: "GitConfigParser", args: Any, *kwargs: Any) -> _T: rval = non_const_func(self, args, *kwargs) self._dirty = True self.write() return rval # END wrapper method flush_changes.__name__ = non_const_func.__name__ return flush_changes class SectionConstraint(Generic[T_ConfigParser]): """Constrains a ConfigParser to only option commands which are constrained to always use the section we have been initialized with. It supports all ConfigParser methods that operate on an option. :note: If used as a context manager, will release the wrapped ConfigParser.""" __slots__ = ("_config", "_section_name") _valid_attrs_ = ( "get_value", "set_value", "get", "set", "getint", "getfloat", "getboolean", "has_option", "remove_section", "remove_option", "options", ) def __init__(self, config: T_ConfigParser, section: str) -> None: self._config = config self._section_name = section def __del__(self) -> None: # Yes, for some reason, we have to call it explicitly for it to work in PY3 ! # Apparently __del__ doesn't get call anymore if refcount becomes 0 # Ridiculous ... . self._config.release() def __getattr__(self, attr: str) -> Any: if attr in self._valid_attrs_: return lambda args, *kwargs: self._call_config(attr, args, *kwargs) return super(SectionConstraint, self).__getattribute__(attr) def _call_config(self, method: str, args: Any, *kwargs: Any) -> Any: """Call the configuration at the given method which must take a section name as first argument""" return getattr(self._config, method)(self._section_name, args, *kwargs) @property def config(self) -> T_ConfigParser: """return: Configparser instance we constrain""" return self._config def release(self) -> None: """Equivalent to GitConfigParser.release(), which is called on our underlying parser instance""" return self._config.release() def __enter__(self) -> "SectionConstraint[T_ConfigParser]": self._config.__enter__() return self def __exit__(self, exception_type: str, exception_value: str, traceback: str) -> None: self._config.__exit__(exception_type, exception_value, traceback) class _OMD(OrderedDict_OMD): """Ordered multi-dict.""" def __setitem__(self, key: str, value: _T) -> None: super(_OMD, self).__setitem__(key, [value]) def add(self, key: str, value: Any) -> None: if key not in self: super(_OMD, self).__setitem__(key, [value]) return None super(_OMD, self).__getitem__(key).append(value) def setall(self, key: str, values: List[_T]) -> None: super(_OMD, self).__setitem__(key, values) def __getitem__(self, key: str) -> Any: return super(_OMD, self).__getitem__(key)[-1] def getlast(self, key: str) -> Any: return super(_OMD, self).__getitem__(key)[-1] def setlast(self, key: str, value: Any) -> None: if key not in self: super(_OMD, self).__setitem__(key, [value]) return prior = super(_OMD, self).__getitem__(key) prior[-1] = value def get(self, key: str, default: Union[_T, None] = None) -> Union[_T, None]: return super(_OMD, self).get(key, [default])[-1] def getall(self, key: str) -> List[_T]: return super(_OMD, self).__getitem__(key) def items(self) -> List[Tuple[str, _T]]: # type: ignore[override] """List of (key, last value for key).""" return [(k, self[k]) for k in self] def items_all(self) -> List[Tuple[str, List[_T]]]: """List of (key, list of values for key).""" return [(k, self.getall(k)) for k in self] def get_config_path(config_level: Lit_config_levels) -> str: # we do not support an absolute path of the gitconfig on windows , # use the global config instead if is_win and config_level == "system": config_level = "global" if config_level == "system": return "/etc/gitconfig" elif config_level == "user": config_home = os.environ.get("XDG_CONFIG_HOME") or osp.join(os.environ.get("HOME", "~"), ".config") return osp.normpath(osp.expanduser(osp.join(config_home, "git", "config"))) elif config_level == "global": return osp.normpath(osp.expanduser("~/.gitconfig")) elif config_level == "repository": raise ValueError("No repo to get repository configuration from. Use Repo._get_config_path") else: # Should not reach here. Will raise ValueError if does. Static typing will warn missing elifs assert_never( config_level, # type: ignore[unreachable] ValueError(f"Invalid configuration level: {config_level!r}"), ) class GitConfigParser(cp.RawConfigParser, metaclass=MetaParserBuilder): """Implements specifics required to read git style configuration files. This variation behaves much like the git.config command such that the configuration will be read on demand based on the filepath given during initialization. The changes will automatically be written once the instance goes out of scope, but can be triggered manually as well. The configuration file will be locked if you intend to change values preventing other instances to write concurrently. :note: The config is case-sensitive even when queried, hence section and option names must match perfectly. If used as a context manager, will release the locked file.""" # { Configuration # The lock type determines the type of lock to use in new configuration readers. # They must be compatible to the LockFile interface. # A suitable alternative would be the BlockingLockFile t_lock = LockFile re_comment = re.compile(r"^\s[#;]") # } END configuration optvalueonly_source = r"\s(?P<option>[^:=\s][^:=])" OPTVALUEONLY = re.compile(optvalueonly_source) OPTCRE = re.compile(optvalueonly_source + r"\s(?P<vi>[:=])\s" + r"(?P<value>.)$") del optvalueonly_source # list of RawConfigParser methods able to change the instance _mutating_methods_ = ("add_section", "remove_section", "remove_option", "set") def __init__( self, file_or_files: Union[None, PathLike, "BytesIO", Sequence[Union[PathLike, "BytesIO"]]] = None, read_only: bool = True, merge_includes: bool = True, config_level: Union[Lit_config_levels, None] = None, repo: Union["Repo", None] = None, ) -> None: """Initialize a configuration reader to read the given file_or_files and to possibly allow changes to it by setting read_only False :param file_or_files: A single file path or file objects or multiple of these :param read_only: If True, the ConfigParser may only read the data , but not change it. If False, only a single file path or file object may be given. We will write back the changes when they happen, or when the ConfigParser is released. This will not happen if other configuration files have been included :param merge_includes: if True, we will read files mentioned in [include] sections and merge their contents into ours. This makes it impossible to write back an individual configuration file. Thus, if you want to modify a single configuration file, turn this off to leave the original dataset unaltered when reading it. :param repo: Reference to repository to use if [includeIf] sections are found in configuration files. """ cp.RawConfigParser.__init__(self, dict_type=_OMD) self._dict: Callable[..., _OMD] # type: ignore # mypy/typeshed bug? self._defaults: _OMD self._sections: _OMD # type: ignore # mypy/typeshed bug? # Used in python 3, needs to stay in sync with sections for underlying implementation to work if not hasattr(self, "_proxies"): self._proxies = self._dict() if file_or_files is not None: self._file_or_files: Union[PathLike, "BytesIO", Sequence[Union[PathLike, "BytesIO"]]] = file_or_files else: if config_level is None: if read_only: self._file_or_files = [ get_config_path(cast(Lit_config_levels, f)) for f in CONFIG_LEVELS if f != "repository" ] else: raise ValueError("No configuration level or configuration files specified") else: self._file_or_files = [get_config_path(config_level)] self._read_only = read_only self._dirty = False self._is_initialized = False self._merge_includes = merge_includes self._repo = repo self._lock: Union["LockFile", None] = None self._acquire_lock() def _acquire_lock(self) -> None: if not self._read_only: if not self._lock: if isinstance(self._file_or_files, (str, os.PathLike)): file_or_files = self._file_or_files elif isinstance(self._file_or_files, (tuple, list, Sequence)): raise ValueError( "Write-ConfigParsers can operate on a single file only, multiple files have been passed" ) else: file_or_files = self._file_or_files.name # END get filename from handle/stream # initialize lock base - we want to write self._lock = self.t_lock(file_or_files) # END lock check self._lock._obtain_lock() # END read-only check def __del__(self) -> None: """Write pending changes if required and release locks""" # NOTE: only consistent in PY2 self.release() def __enter__(self) -> "GitConfigParser": self._acquire_lock() return self def __exit__(self, args: Any) -> None: self.release() def release(self) -> None: """Flush changes and release the configuration write lock. This instance must not be used anymore afterwards. In Python 3, it's required to explicitly release locks and flush changes, as __del__ is not called deterministically anymore.""" # checking for the lock here makes sure we do not raise during write() # in case an invalid parser was created who could not get a lock if self.read_only or (self._lock and not self._lock._has_lock()): return try: try: self.write() except IOError: log.error("Exception during destruction of GitConfigParser", exc_info=True) except ReferenceError: # This happens in PY3 ... and usually means that some state cannot be written # as the sections dict cannot be iterated # Usually when shutting down the interpreter, don'y know how to fix this pass finally: if self._lock is not None: self._lock._release_lock() def optionxform(self, optionstr: str) -> str: """Do not transform options in any way when writing""" return optionstr def _read(self, fp: Union[BufferedReader, IO[bytes]], fpname: str) -> None: """A direct copy of the py2.4 version of the super class's _read method to assure it uses ordered dicts. Had to change one line to make it work. Future versions have this fixed, but in fact its quite embarrassing for the guys not to have done it right in the first place ! Removed big comments to make it more compact. Made sure it ignores initial whitespace as git uses tabs""" cursect = None # None, or a dictionary optname = None lineno = 0 is_multi_line = False e = None # None, or an exception def string_decode(v: str) -> str: if v[-1] == "\\": v = v[:-1] # end cut trailing escapes to prevent decode error return v.encode(defenc).decode("unicode_escape") # end # end while True: # we assume to read binary ! line = fp.readline().decode(defenc) if not line: break lineno = lineno + 1 # comment or blank line? if line.strip() == "" or self.re_comment.match(line): continue if line.split(None, 1)[0].lower() == "rem" and line[0] in "rR": # no leading whitespace continue # is it a section header? mo = self.SECTCRE.match(line.strip()) if not is_multi_line and mo: sectname: str = mo.group("header").strip() if sectname in self._sections: cursect = self._sections[sectname] elif sectname == cp.DEFAULTSECT: cursect = self._defaults else: cursect = self._dict((("__name__", sectname),)) self._sections[sectname] = cursect self._proxies[sectname] = None # So sections can't start with a continuation line optname = None # no section header in the file? elif cursect is None: raise cp.MissingSectionHeaderError(fpname, lineno, line) # an option line? elif not is_multi_line: mo = self.OPTCRE.match(line) if mo: # We might just have handled the last line, which could contain a quotation we want to remove optname, vi, optval = mo.group("option", "vi", "value") if vi in ("=", ":") and ";" in optval and not optval.strip().startswith('"'): pos = optval.find(";") if pos != -1 and optval[pos - 1].isspace(): optval = optval[:pos] optval = optval.strip() if optval == '""': optval = "" # end handle empty string optname = self.optionxform(optname.rstrip()) if len(optval) > 1 and optval[0] == '"' and optval[-1] != '"': is_multi_line = True optval = string_decode(optval[1:]) # end handle multi-line # preserves multiple values for duplicate optnames cursect.add(optname, optval) else: # check if it's an option with no value - it's just ignored by git if not self.OPTVALUEONLY.match(line): if not e: e = cp.ParsingError(fpname) e.append(lineno, repr(line)) continue else: line = line.rstrip() if line.endswith('"'): is_multi_line = False line = line[:-1] # end handle quotations optval = cursect.getlast(optname) cursect.setlast(optname, optval + string_decode(line)) # END parse section or option # END while reading # if any parsing errors occurred, raise an exception if e: raise e def _has_includes(self) -> Union[bool, int]: return self._merge_includes and len(self._included_paths()) def _included_paths(self) -> List[Tuple[str, str]]: """Return List all paths that must be included to configuration as Tuples of (option, value). """ paths = [] for section in self.sections(): if section == "include": paths += self.items(section) match = CONDITIONAL_INCLUDE_REGEXP.search(section) if match is None or self._repo is None: continue keyword = match.group(1) value = match.group(2).strip() if keyword in ["gitdir", "gitdir/i"]: value = osp.expanduser(value) if not any(value.startswith(s) for s in ["./", "/"]): value = "/" + value if value.endswith("/"): value += "" # Ensure that glob is always case insensitive if required. if keyword.endswith("/i"): value = re.sub( r"[a-zA-Z]", lambda m: "[{}{}]".format(m.group().lower(), m.group().upper()), value, ) if self._repo.git_dir: if fnmatch.fnmatchcase(str(self._repo.git_dir), value): paths += self.items(section) elif keyword == "onbranch": try: branch_name = self._repo.active_branch.name except TypeError: # Ignore section if active branch cannot be retrieved. continue if fnmatch.fnmatchcase(branch_name, value): paths += self.items(section) return paths def read(self) -> None: # type: ignore[override] """Reads the data stored in the files we have been initialized with. It will ignore files that cannot be read, possibly leaving an empty configuration :return: Nothing :raise IOError: if a file cannot be handled""" if self._is_initialized: return None self._is_initialized = True files_to_read: List[Union[PathLike, IO]] = [""] if isinstance(self._file_or_files, (str, os.PathLike)): # for str or Path, as str is a type of Sequence files_to_read = [self._file_or_files] elif not isinstance(self._file_or_files, (tuple, list, Sequence)): # could merge with above isinstance once runtime type known files_to_read = [self._file_or_files] else: # for lists or tuples files_to_read = list(self._file_or_files) # end assure we have a copy of the paths to handle seen = set(files_to_read) num_read_include_files = 0 while files_to_read: file_path = files_to_read.pop(0) file_ok = False if hasattr(file_path, "seek"): # must be a file objectfile-object file_path = cast(IO[bytes], file_path) # replace with assert to narrow type, once sure self._read(file_path, file_path.name) else: # assume a path if it is not a file-object file_path = cast(PathLike, file_path) try: with open(file_path, "rb") as fp: file_ok = True self._read(fp, fp.name) except IOError: continue # Read includes and append those that we didn't handle yet # We expect all paths to be normalized and absolute (and will assure that is the case) if self._has_includes(): for _, include_path in self._included_paths(): if include_path.startswith("~"): include_path = osp.expanduser(include_path) if not osp.isabs(include_path): if not file_ok: continue # end ignore relative paths if we don't know the configuration file path file_path = cast(PathLike, file_path) assert osp.isabs(file_path), "Need absolute paths to be sure our cycle checks will work" include_path = osp.join(osp.dirname(file_path), include_path) # end make include path absolute include_path = osp.normpath(include_path) if include_path in seen or not os.access(include_path, os.R_OK): continue seen.add(include_path) # insert included file to the top to be considered first files_to_read.insert(0, include_path) num_read_include_files += 1 # each include path in configuration file # end handle includes # END for each file object to read # If there was no file included, we can safely write back (potentially) the configuration file # without altering it's meaning if num_read_include_files == 0: self._merge_includes = False # end def _write(self, fp: IO) -> None: """Write an .ini-format representation of the configuration state in git compatible format""" def write_section(name: str, section_dict: _OMD) -> None: fp.write(("[%s]\n" % name).encode(defenc)) values: Sequence[str] # runtime only gets str in tests, but should be whatever _OMD stores v: str for (key, values) in section_dict.items_all(): if key == "__name__": continue for v in values: fp.write(("\t%s = %s\n" % (key, self._value_to_string(v).replace("\n", "\n\t"))).encode(defenc)) # END if key is not __name__ # END section writing if self._defaults: write_section(cp.DEFAULTSECT, self._defaults) value: _OMD for name, value in self._sections.items(): write_section(name, value) def items(self, section_name: str) -> List[Tuple[str, str]]: # type: ignore[override] """:return: list((option, value), ...) pairs of all items in the given section""" return [(k, v) for k, v in super(GitConfigParser, self).items(section_name) if k != "__name__"] def items_all(self, section_name: str) -> List[Tuple[str, List[str]]]: """:return: list((option, [values...]), ...) pairs of all items in the given section""" rv = _OMD(self._defaults) for k, vs in self._sections[section_name].items_all(): if k == "__name__": continue if k in rv and rv.getall(k) == vs: continue for v in vs: rv.add(k, v) return rv.items_all() @needs_values def write(self) -> None: """Write changes to our file, if there are changes at all :raise IOError: if this is a read-only writer instance or if we could not obtain a file lock""" self._assure_writable("write") if not self._dirty: return None if isinstance(self._file_or_files, (list, tuple)): raise AssertionError( "Cannot write back if there is not exactly a single file to write to, have %i files" % len(self._file_or_files) ) # end assert multiple files if self._has_includes(): log.debug( "Skipping write-back of configuration file as include files were merged in." + "Set merge_includes=False to prevent this." ) return None # end fp = self._file_or_files # we have a physical file on disk, so get a lock is_file_lock = isinstance(fp, (str, os.PathLike, IOBase)) # can't use Pathlike until 3.5 dropped if is_file_lock and self._lock is not None: # else raise Error? self._lock._obtain_lock() if not hasattr(fp, "seek"): fp = cast(PathLike, fp) with open(fp, "wb") as fp_open: self._write(fp_open) else: fp = cast("BytesIO", fp) fp.seek(0) # make sure we do not overwrite into an existing file if hasattr(fp, "truncate"): fp.truncate() self._write(fp) def _assure_writable(self, method_name: str) -> None: if self.read_only: raise IOError("Cannot execute non-constant method %s.%s" % (self, method_name)) def add_section(self, section: str) -> None: """Assures added options will stay in order""" return super(GitConfigParser, self).add_section(section) @property def read_only(self) -> bool: """:return: True if this instance may change the configuration file""" return self._read_only def get_value( self, section: str, option: str, default: Union[int, float, str, bool, None] = None, ) -> Union[int, float, str, bool]: # can default or return type include bool? """Get an option's value. If multiple values are specified for this option in the section, the last one specified is returned. :param default: If not None, the given default value will be returned in case the option did not exist :return: a properly typed value, either int, float or string :raise TypeError: in case the value could not be understood Otherwise the exceptions known to the ConfigParser will be raised.""" try: valuestr = self.get(section, option) except Exception: if default is not None: return default raise return self._string_to_value(valuestr) def get_values( self, section: str, option: str, default: Union[int, float, str, bool, None] = None, ) -> List[Union[int, float, str, bool]]: """Get an option's values. If multiple values are specified for this option in the section, all are returned. :param default: If not None, a list containing the given default value will be returned in case the option did not exist :return: a list of properly typed values, either int, float or string :raise TypeError: in case the value could not be understood Otherwise the exceptions known to the ConfigParser will be raised.""" try: self.sections() lst = self._sections[section].getall(option) except Exception: if default is not None: return [default] raise return [self._string_to_value(valuestr) for valuestr in lst] def _string_to_value(self, valuestr: str) -> Union[int, float, str, bool]: types = (int, float) for numtype in types: try: val = numtype(valuestr) # truncated value ? if val != float(valuestr): continue return val except (ValueError, TypeError): continue # END for each numeric type # try boolean values as git uses them vl = valuestr.lower() if vl == "false": return False if vl == "true": return True if not isinstance(valuestr, str): raise TypeError( "Invalid value type: only int, long, float and str are allowed", valuestr, ) return valuestr def _value_to_string(self, value: Union[str, bytes, int, float, bool]) -> str: if isinstance(value, (int, float, bool)): return str(value) return force_text(value) @needs_values @set_dirty_and_flush_changes def set_value(self, section: str, option: str, value: Union[str, bytes, int, float, bool]) -> "GitConfigParser": """Sets the given option in section to the given value. It will create the section if required, and will not throw as opposed to the default ConfigParser 'set' method. :param section: Name of the section in which the option resides or should reside :param option: Name of the options whose value to set :param value: Value to set the option to. It must be a string or convertible to a string :return: this instance""" if not self.has_section(section): self.add_section(section) self.set(section, option, self._value_to_string(value)) return self @needs_values @set_dirty_and_flush_changes def add_value(self, section: str, option: str, value: Union[str, bytes, int, float, bool]) -> "GitConfigParser": """Adds a value for the given option in section. It will create the section if required, and will not throw as opposed to the default ConfigParser 'set' method. The value becomes the new value of the option as returned by 'get_value', and appends to the list of values returned by 'get_values`'. :param section: Name of the section in which the option resides or should reside :param option: Name of the option :param value: Value to add to option. It must be a string or convertible to a string :return: this instance""" if not self.has_section(section): self.add_section(section) self._sections[section].add(option, self._value_to_string(value)) return self def rename_section(self, section: str, new_name: str) -> "GitConfigParser": """rename the given section to new_name :raise ValueError: if section doesn't exit :raise ValueError: if a section with new_name does already exist :return: this instance """ if not self.has_section(section): raise ValueError("Source section '%s' doesn't exist" % section) if self.has_section(new_name): raise ValueError("Destination section '%s' already exists" % new_name) super(GitConfigParser, self).add_section(new_name) new_section = self._sections[new_name] for k, vs in self.items_all(section): new_section.setall(k, vs) # end for each value to copy # This call writes back the changes, which is why we don't have the respective decorator self.remove_section(section) return self	33.77 KB	2026-01-20 12:58
📄 db.py """Module with our own gitdb implementation - it uses the git command""" from git.util import bin_to_hex, hex_to_bin from gitdb.base import OInfo, OStream from gitdb.db import GitDB # @UnusedImport from gitdb.db import LooseObjectDB from gitdb.exc import BadObject from git.exc import GitCommandError # typing------------------------------------------------- from typing import TYPE_CHECKING from git.types import PathLike if TYPE_CHECKING: from git.cmd import Git # -------------------------------------------------------- __all__ = ("GitCmdObjectDB", "GitDB") class GitCmdObjectDB(LooseObjectDB): """A database representing the default git object store, which includes loose objects, pack files and an alternates file It will create objects only in the loose object database. :note: for now, we use the git command to do all the lookup, just until he have packs and the other implementations """ def __init__(self, root_path: PathLike, git: "Git") -> None: """Initialize this instance with the root and a git command""" super(GitCmdObjectDB, self).__init__(root_path) self._git = git def info(self, binsha: bytes) -> OInfo: hexsha, typename, size = self._git.get_object_header(bin_to_hex(binsha)) return OInfo(hex_to_bin(hexsha), typename, size) def stream(self, binsha: bytes) -> OStream: """For now, all lookup is done by git itself""" hexsha, typename, size, stream = self._git.stream_object_data(bin_to_hex(binsha)) return OStream(hex_to_bin(hexsha), typename, size, stream) # { Interface def partial_to_complete_sha_hex(self, partial_hexsha: str) -> bytes: """:return: Full binary 20 byte sha from the given partial hexsha :raise AmbiguousObjectName: :raise BadObject: :note: currently we only raise BadObject as git does not communicate AmbiguousObjects separately""" try: hexsha, _typename, _size = self._git.get_object_header(partial_hexsha) return hex_to_bin(hexsha) except (GitCommandError, ValueError) as e: raise BadObject(partial_hexsha) from e # END handle exceptions # } END interface	2.19 KB	2026-01-20 12:58
📄 diff.py # diff.py # Copyright (C) 2008, 2009 Michael Trier (mtrier@gmail.com) and contributors # # This module is part of GitPython and is released under # the BSD License: http://www.opensource.org/licenses/bsd-license.php import re from git.cmd import handle_process_output from git.compat import defenc from git.util import finalize_process, hex_to_bin from .objects.blob import Blob from .objects.util import mode_str_to_int # typing ------------------------------------------------------------------ from typing import ( Any, Iterator, List, Match, Optional, Tuple, Type, TypeVar, Union, TYPE_CHECKING, cast, ) from git.types import PathLike, Literal if TYPE_CHECKING: from .objects.tree import Tree from .objects import Commit from git.repo.base import Repo from git.objects.base import IndexObject from subprocess import Popen from git import Git Lit_change_type = Literal["A", "D", "C", "M", "R", "T", "U"] # def is_change_type(inp: str) -> TypeGuard[Lit_change_type]: # # return True # return inp in ['A', 'D', 'C', 'M', 'R', 'T', 'U'] # ------------------------------------------------------------------------ __all__ = ("Diffable", "DiffIndex", "Diff", "NULL_TREE") # Special object to compare against the empty tree in diffs NULL_TREE = object() _octal_byte_re = re.compile(b"\\\\([0-9]{3})") def _octal_repl(matchobj: Match) -> bytes: value = matchobj.group(1) value = int(value, 8) value = bytes(bytearray((value,))) return value def decode_path(path: bytes, has_ab_prefix: bool = True) -> Optional[bytes]: if path == b"/dev/null": return None if path.startswith(b'"') and path.endswith(b'"'): path = path[1:-1].replace(b"\\n", b"\n").replace(b"\\t", b"\t").replace(b'\\"', b'"').replace(b"\\\\", b"\\") path = _octal_byte_re.sub(_octal_repl, path) if has_ab_prefix: assert path.startswith(b"a/") or path.startswith(b"b/") path = path[2:] return path class Diffable(object): """Common interface for all object that can be diffed against another object of compatible type. :note: Subclasses require a repo member as it is the case for Object instances, for practical reasons we do not derive from Object.""" __slots__ = () # standin indicating you want to diff against the index class Index(object): pass def _process_diff_args( self, args: List[Union[str, "Diffable", Type["Diffable.Index"], object]] ) -> List[Union[str, "Diffable", Type["Diffable.Index"], object]]: """ :return: possibly altered version of the given args list. Method is called right before git command execution. Subclasses can use it to alter the behaviour of the superclass""" return args def diff( self, other: Union[Type["Index"], "Tree", "Commit", None, str, object] = Index, paths: Union[PathLike, List[PathLike], Tuple[PathLike, ...], None] = None, create_patch: bool = False, *kwargs: Any, ) -> "DiffIndex": """Creates diffs between two items being trees, trees and index or an index and the working tree. It will detect renames automatically. :param other: Is the item to compare us with. If None, we will be compared to the working tree. If Treeish, it will be compared against the respective tree If Index ( type ), it will be compared against the index. If git.NULL_TREE, it will compare against the empty tree. It defaults to Index to assure the method will not by-default fail on bare repositories. :param paths: is a list of paths or a single path to limit the diff to. It will only include at least one of the given path or paths. :param create_patch: If True, the returned Diff contains a detailed patch that if applied makes the self to other. Patches are somewhat costly as blobs have to be read and diffed. :param kwargs: Additional arguments passed to git-diff, such as R=True to swap both sides of the diff. :return: git.DiffIndex :note: On a bare repository, 'other' needs to be provided as Index or as as Tree/Commit, or a git command error will occur""" args: List[Union[PathLike, Diffable, Type["Diffable.Index"], object]] = [] args.append("--abbrev=40") # we need full shas args.append("--full-index") # get full index paths, not only filenames # remove default '-M' arg (check for renames) if user is overriding it if not any(x in kwargs for x in ('find_renames', 'no_renames', 'M')): args.append("-M") if create_patch: args.append("-p") else: args.append("--raw") args.append("-z") # in any way, assure we don't see colored output, # fixes https://github.com/gitpython-developers/GitPython/issues/172 args.append("--no-color") if paths is not None and not isinstance(paths, (tuple, list)): paths = [paths] if hasattr(self, "Has_Repo"): self.repo: "Repo" = self.repo diff_cmd = self.repo.git.diff if other is self.Index: args.insert(0, "--cached") elif other is NULL_TREE: args.insert(0, "-r") # recursive diff-tree args.insert(0, "--root") diff_cmd = self.repo.git.diff_tree elif other is not None: args.insert(0, "-r") # recursive diff-tree args.insert(0, other) diff_cmd = self.repo.git.diff_tree args.insert(0, self) # paths is list here or None if paths: args.append("--") args.extend(paths) # END paths handling kwargs["as_process"] = True proc = diff_cmd(self._process_diff_args(args), *kwargs) diff_method = Diff._index_from_patch_format if create_patch else Diff._index_from_raw_format index = diff_method(self.repo, proc) proc.wait() return index T_Diff = TypeVar("T_Diff", bound="Diff") class DiffIndex(List[T_Diff]): """Implements an Index for diffs, allowing a list of Diffs to be queried by the diff properties. The class improves the diff handling convenience""" # change type invariant identifying possible ways a blob can have changed # A = Added # D = Deleted # R = Renamed # M = Modified # T = Changed in the type change_type = ("A", "C", "D", "R", "M", "T") def iter_change_type(self, change_type: Lit_change_type) -> Iterator[T_Diff]: """ :return: iterator yielding Diff instances that match the given change_type :param change_type: Member of DiffIndex.change_type, namely: 'A' for added paths * 'D' for deleted paths * 'R' for renamed paths * 'M' for paths with modified data * 'T' for changed in the type paths """ if change_type not in self.change_type: raise ValueError("Invalid change type: %s" % change_type) for diffidx in self: if diffidx.change_type == change_type: yield diffidx elif change_type == "A" and diffidx.new_file: yield diffidx elif change_type == "D" and diffidx.deleted_file: yield diffidx elif change_type == "C" and diffidx.copied_file: yield diffidx elif change_type == "R" and diffidx.renamed: yield diffidx elif change_type == "M" and diffidx.a_blob and diffidx.b_blob and diffidx.a_blob != diffidx.b_blob: yield diffidx # END for each diff class Diff(object): """A Diff contains diff information between two Trees. It contains two sides a and b of the diff, members are prefixed with "a" and "b" respectively to inidcate that. Diffs keep information about the changed blob objects, the file mode, renames, deletions and new files. There are a few cases where None has to be expected as member variable value: ``New File``:: a_mode is None a_blob is None a_path is None ``Deleted File``:: b_mode is None b_blob is None b_path is None ``Working Tree Blobs`` When comparing to working trees, the working tree blob will have a null hexsha as a corresponding object does not yet exist. The mode will be null as well. But the path will be available though. If it is listed in a diff the working tree version of the file must be different to the version in the index or tree, and hence has been modified.""" # precompiled regex re_header = re.compile( rb""" ^diff[ ]--git [ ](?P<a_path_fallback>"?[ab]/.+?"?)[ ](?P<b_path_fallback>"?[ab]/.+?"?)\n (?:^old[ ]mode[ ](?P<old_mode>\d+)\n ^new[ ]mode[ ](?P<new_mode>\d+)(?:\n\|$))? (?:^similarity[ ]index[ ]\d+%\n ^rename[ ]from[ ](?P<rename_from>.)\n ^rename[ ]to[ ](?P<rename_to>.)(?:\n\|$))? (?:^new[ ]file[ ]mode[ ](?P<new_file_mode>.+)(?:\n\|$))? (?:^deleted[ ]file[ ]mode[ ](?P<deleted_file_mode>.+)(?:\n\|$))? (?:^similarity[ ]index[ ]\d+%\n ^copy[ ]from[ ].\n ^copy[ ]to[ ](?P<copied_file_name>.)(?:\n\|$))? (?:^index[ ](?P<a_blob_id>[0-9A-Fa-f]+) \.\.(?P<b_blob_id>[0-9A-Fa-f]+)[ ]?(?P<b_mode>.+)?(?:\n\|$))? (?:^---[ ](?P<a_path>[^\t\n\r\f\v])[\t\r\f\v](?:\n\|$))? (?:^\+\+\+[ ](?P<b_path>[^\t\n\r\f\v])[\t\r\f\v](?:\n\|$))? """, re.VERBOSE \| re.MULTILINE, ) # can be used for comparisons NULL_HEX_SHA = "0" * 40 NULL_BIN_SHA = b"\0" * 20 __slots__ = ( "a_blob", "b_blob", "a_mode", "b_mode", "a_rawpath", "b_rawpath", "new_file", "deleted_file", "copied_file", "raw_rename_from", "raw_rename_to", "diff", "change_type", "score", ) def __init__( self, repo: "Repo", a_rawpath: Optional[bytes], b_rawpath: Optional[bytes], a_blob_id: Union[str, bytes, None], b_blob_id: Union[str, bytes, None], a_mode: Union[bytes, str, None], b_mode: Union[bytes, str, None], new_file: bool, deleted_file: bool, copied_file: bool, raw_rename_from: Optional[bytes], raw_rename_to: Optional[bytes], diff: Union[str, bytes, None], change_type: Optional[Lit_change_type], score: Optional[int], ) -> None: assert a_rawpath is None or isinstance(a_rawpath, bytes) assert b_rawpath is None or isinstance(b_rawpath, bytes) self.a_rawpath = a_rawpath self.b_rawpath = b_rawpath self.a_mode = mode_str_to_int(a_mode) if a_mode else None self.b_mode = mode_str_to_int(b_mode) if b_mode else None # Determine whether this diff references a submodule, if it does then # we need to overwrite "repo" to the corresponding submodule's repo instead if repo and a_rawpath: for submodule in repo.submodules: if submodule.path == a_rawpath.decode(defenc, "replace"): if submodule.module_exists(): repo = submodule.module() break self.a_blob: Union["IndexObject", None] if a_blob_id is None or a_blob_id == self.NULL_HEX_SHA: self.a_blob = None else: self.a_blob = Blob(repo, hex_to_bin(a_blob_id), mode=self.a_mode, path=self.a_path) self.b_blob: Union["IndexObject", None] if b_blob_id is None or b_blob_id == self.NULL_HEX_SHA: self.b_blob = None else: self.b_blob = Blob(repo, hex_to_bin(b_blob_id), mode=self.b_mode, path=self.b_path) self.new_file: bool = new_file self.deleted_file: bool = deleted_file self.copied_file: bool = copied_file # be clear and use None instead of empty strings assert raw_rename_from is None or isinstance(raw_rename_from, bytes) assert raw_rename_to is None or isinstance(raw_rename_to, bytes) self.raw_rename_from = raw_rename_from or None self.raw_rename_to = raw_rename_to or None self.diff = diff self.change_type: Union[Lit_change_type, None] = change_type self.score = score def __eq__(self, other: object) -> bool: for name in self.__slots__: if getattr(self, name) != getattr(other, name): return False # END for each name return True def __ne__(self, other: object) -> bool: return not (self == other) def __hash__(self) -> int: return hash(tuple(getattr(self, n) for n in self.__slots__)) def __str__(self) -> str: h: str = "%s" if self.a_blob: h %= self.a_blob.path elif self.b_blob: h %= self.b_blob.path msg: str = "" line = None # temp line line_length = 0 # line length for b, n in zip((self.a_blob, self.b_blob), ("lhs", "rhs")): if b: line = "\n%s: %o \| %s" % (n, b.mode, b.hexsha) else: line = "\n%s: None" % n # END if blob is not None line_length = max(len(line), line_length) msg += line # END for each blob # add headline h += "\n" + "=" * line_length if self.deleted_file: msg += "\nfile deleted in rhs" if self.new_file: msg += "\nfile added in rhs" if self.copied_file: msg += "\nfile %r copied from %r" % (self.b_path, self.a_path) if self.rename_from: msg += "\nfile renamed from %r" % self.rename_from if self.rename_to: msg += "\nfile renamed to %r" % self.rename_to if self.diff: msg += "\n---" try: msg += self.diff.decode(defenc) if isinstance(self.diff, bytes) else self.diff except UnicodeDecodeError: msg += "OMITTED BINARY DATA" # end handle encoding msg += "\n---" # END diff info # Python2 silliness: have to assure we convert our likely to be unicode object to a string with the # right encoding. Otherwise it tries to convert it using ascii, which may fail ungracefully res = h + msg # end return res @property def a_path(self) -> Optional[str]: return self.a_rawpath.decode(defenc, "replace") if self.a_rawpath else None @property def b_path(self) -> Optional[str]: return self.b_rawpath.decode(defenc, "replace") if self.b_rawpath else None @property def rename_from(self) -> Optional[str]: return self.raw_rename_from.decode(defenc, "replace") if self.raw_rename_from else None @property def rename_to(self) -> Optional[str]: return self.raw_rename_to.decode(defenc, "replace") if self.raw_rename_to else None @property def renamed(self) -> bool: """:returns: True if the blob of our diff has been renamed :note: This property is deprecated, please use ``renamed_file`` instead. """ return self.renamed_file @property def renamed_file(self) -> bool: """:returns: True if the blob of our diff has been renamed""" return self.rename_from != self.rename_to @classmethod def _pick_best_path(cls, path_match: bytes, rename_match: bytes, path_fallback_match: bytes) -> Optional[bytes]: if path_match: return decode_path(path_match) if rename_match: return decode_path(rename_match, has_ab_prefix=False) if path_fallback_match: return decode_path(path_fallback_match) return None @classmethod def _index_from_patch_format(cls, repo: "Repo", proc: Union["Popen", "Git.AutoInterrupt"]) -> DiffIndex: """Create a new DiffIndex from the given text which must be in patch format :param repo: is the repository we are operating on - it is required :param stream: result of 'git diff' as a stream (supporting file protocol) :return: git.DiffIndex""" ## FIXME: Here SLURPING raw, need to re-phrase header-regexes linewise. text_list: List[bytes] = [] handle_process_output(proc, text_list.append, None, finalize_process, decode_streams=False) # for now, we have to bake the stream text = b"".join(text_list) index: "DiffIndex" = DiffIndex() previous_header: Union[Match[bytes], None] = None header: Union[Match[bytes], None] = None a_path, b_path = None, None # for mypy a_mode, b_mode = None, None # for mypy for _header in cls.re_header.finditer(text): ( a_path_fallback, b_path_fallback, old_mode, new_mode, rename_from, rename_to, new_file_mode, deleted_file_mode, copied_file_name, a_blob_id, b_blob_id, b_mode, a_path, b_path, ) = _header.groups() new_file, deleted_file, copied_file = ( bool(new_file_mode), bool(deleted_file_mode), bool(copied_file_name), ) a_path = cls._pick_best_path(a_path, rename_from, a_path_fallback) b_path = cls._pick_best_path(b_path, rename_to, b_path_fallback) # Our only means to find the actual text is to see what has not been matched by our regex, # and then retro-actively assign it to our index if previous_header is not None: index[-1].diff = text[previous_header.end() : _header.start()] # end assign actual diff # Make sure the mode is set if the path is set. Otherwise the resulting blob is invalid # We just use the one mode we should have parsed a_mode = old_mode or deleted_file_mode or (a_path and (b_mode or new_mode or new_file_mode)) b_mode = b_mode or new_mode or new_file_mode or (b_path and a_mode) index.append( Diff( repo, a_path, b_path, a_blob_id and a_blob_id.decode(defenc), b_blob_id and b_blob_id.decode(defenc), a_mode and a_mode.decode(defenc), b_mode and b_mode.decode(defenc), new_file, deleted_file, copied_file, rename_from, rename_to, None, None, None, ) ) previous_header = _header header = _header # end for each header we parse if index and header: index[-1].diff = text[header.end() :] # end assign last diff return index @staticmethod def _handle_diff_line(lines_bytes: bytes, repo: "Repo", index: DiffIndex) -> None: lines = lines_bytes.decode(defenc) # Discard everything before the first colon, and the colon itself. _, _, lines = lines.partition(":") for line in lines.split("\x00:"): if not line: # The line data is empty, skip continue meta, _, path = line.partition("\x00") path = path.rstrip("\x00") a_blob_id: Optional[str] b_blob_id: Optional[str] old_mode, new_mode, a_blob_id, b_blob_id, _change_type = meta.split(None, 4) # Change type can be R100 # R: status letter # 100: score (in case of copy and rename) # assert is_change_type(_change_type[0]), f"Unexpected value for change_type received: {_change_type[0]}" change_type: Lit_change_type = cast(Lit_change_type, _change_type[0]) score_str = "".join(_change_type[1:]) score = int(score_str) if score_str.isdigit() else None path = path.strip() a_path = path.encode(defenc) b_path = path.encode(defenc) deleted_file = False new_file = False copied_file = False rename_from = None rename_to = None # NOTE: We cannot conclude from the existence of a blob to change type # as diffs with the working do not have blobs yet if change_type == "D": b_blob_id = None # Optional[str] deleted_file = True elif change_type == "A": a_blob_id = None new_file = True elif change_type == "C": copied_file = True a_path_str, b_path_str = path.split("\x00", 1) a_path = a_path_str.encode(defenc) b_path = b_path_str.encode(defenc) elif change_type == "R": a_path_str, b_path_str = path.split("\x00", 1) a_path = a_path_str.encode(defenc) b_path = b_path_str.encode(defenc) rename_from, rename_to = a_path, b_path elif change_type == "T": # Nothing to do pass # END add/remove handling diff = Diff( repo, a_path, b_path, a_blob_id, b_blob_id, old_mode, new_mode, new_file, deleted_file, copied_file, rename_from, rename_to, "", change_type, score, ) index.append(diff) @classmethod def _index_from_raw_format(cls, repo: "Repo", proc: "Popen") -> "DiffIndex": """Create a new DiffIndex from the given stream which must be in raw format. :return: git.DiffIndex""" # handles # :100644 100644 687099101... 37c5e30c8... M .gitignore index: "DiffIndex" = DiffIndex() handle_process_output( proc, lambda byt: cls._handle_diff_line(byt, repo, index), None, finalize_process, decode_streams=False, ) return index	22.88 KB	2026-01-20 12:58
📄 exc.py # exc.py # Copyright (C) 2008, 2009 Michael Trier (mtrier@gmail.com) and contributors # # This module is part of GitPython and is released under # the BSD License: http://www.opensource.org/licenses/bsd-license.php """ Module containing all exceptions thrown throughout the git package, """ from gitdb.exc import BadName # NOQA @UnusedWildImport skipcq: PYL-W0401, PYL-W0614 from gitdb.exc import * # NOQA @UnusedWildImport skipcq: PYL-W0401, PYL-W0614 from git.compat import safe_decode from git.util import remove_password_if_present # typing ---------------------------------------------------- from typing import List, Sequence, Tuple, Union, TYPE_CHECKING from git.types import PathLike if TYPE_CHECKING: from git.repo.base import Repo # ------------------------------------------------------------------ class GitError(Exception): """Base class for all package exceptions""" class InvalidGitRepositoryError(GitError): """Thrown if the given repository appears to have an invalid format.""" class WorkTreeRepositoryUnsupported(InvalidGitRepositoryError): """Thrown to indicate we can't handle work tree repositories""" class NoSuchPathError(GitError, OSError): """Thrown if a path could not be access by the system.""" class UnsafeProtocolError(GitError): """Thrown if unsafe protocols are passed without being explicitly allowed.""" class UnsafeOptionError(GitError): """Thrown if unsafe options are passed without being explicitly allowed.""" class CommandError(GitError): """Base class for exceptions thrown at every stage of `Popen()` execution. :param command: A non-empty list of argv comprising the command-line. """ #: A unicode print-format with 2 `%s for `<cmdline>` and the rest, #: e.g. #: "'%s' failed%s" _msg = "Cmd('%s') failed%s" def __init__( self, command: Union[List[str], Tuple[str, ...], str], status: Union[str, int, None, Exception] = None, stderr: Union[bytes, str, None] = None, stdout: Union[bytes, str, None] = None, ) -> None: if not isinstance(command, (tuple, list)): command = command.split() self.command = remove_password_if_present(command) self.status = status if status: if isinstance(status, Exception): status = "%s('%s')" % (type(status).__name__, safe_decode(str(status))) else: try: status = "exit code(%s)" % int(status) except (ValueError, TypeError): s = safe_decode(str(status)) status = "'%s'" % s if isinstance(status, str) else s self._cmd = safe_decode(self.command[0]) self._cmdline = " ".join(safe_decode(i) for i in self.command) self._cause = status and " due to: %s" % status or "!" stdout_decode = safe_decode(stdout) stderr_decode = safe_decode(stderr) self.stdout = stdout_decode and "\n stdout: '%s'" % stdout_decode or "" self.stderr = stderr_decode and "\n stderr: '%s'" % stderr_decode or "" def __str__(self) -> str: return (self._msg + "\n cmdline: %s%s%s") % ( self._cmd, self._cause, self._cmdline, self.stdout, self.stderr, ) class GitCommandNotFound(CommandError): """Thrown if we cannot find the `git` executable in the PATH or at the path given by the GIT_PYTHON_GIT_EXECUTABLE environment variable""" def __init__(self, command: Union[List[str], Tuple[str], str], cause: Union[str, Exception]) -> None: super(GitCommandNotFound, self).__init__(command, cause) self._msg = "Cmd('%s') not found%s" class GitCommandError(CommandError): """Thrown if execution of the git command fails with non-zero status code.""" def __init__( self, command: Union[List[str], Tuple[str, ...], str], status: Union[str, int, None, Exception] = None, stderr: Union[bytes, str, None] = None, stdout: Union[bytes, str, None] = None, ) -> None: super(GitCommandError, self).__init__(command, status, stderr, stdout) class CheckoutError(GitError): """Thrown if a file could not be checked out from the index as it contained changes. The .failed_files attribute contains a list of relative paths that failed to be checked out as they contained changes that did not exist in the index. The .failed_reasons attribute contains a string informing about the actual cause of the issue. The .valid_files attribute contains a list of relative paths to files that were checked out successfully and hence match the version stored in the index""" def __init__( self, message: str, failed_files: Sequence[PathLike], valid_files: Sequence[PathLike], failed_reasons: List[str], ) -> None: Exception.__init__(self, message) self.failed_files = failed_files self.failed_reasons = failed_reasons self.valid_files = valid_files def __str__(self) -> str: return Exception.__str__(self) + ":%s" % self.failed_files class CacheError(GitError): """Base for all errors related to the git index, which is called cache internally""" class UnmergedEntriesError(CacheError): """Thrown if an operation cannot proceed as there are still unmerged entries in the cache""" class HookExecutionError(CommandError): """Thrown if a hook exits with a non-zero exit code. It provides access to the exit code and the string returned via standard output""" def __init__( self, command: Union[List[str], Tuple[str, ...], str], status: Union[str, int, None, Exception], stderr: Union[bytes, str, None] = None, stdout: Union[bytes, str, None] = None, ) -> None: super(HookExecutionError, self).__init__(command, status, stderr, stdout) self._msg = "Hook('%s') failed%s" class RepositoryDirtyError(GitError): """Thrown whenever an operation on a repository fails as it has uncommitted changes that would be overwritten""" def __init__(self, repo: "Repo", message: str) -> None: self.repo = repo self.message = message def __str__(self) -> str: return "Operation cannot be performed on %r: %s" % (self.repo, self.message)	6.29 KB	2026-01-20 12:58
📁 index	-	2026-03-21 15:45
📁 objects	-	2026-03-21 15:45
📄 py.typed	0 B	2026-01-20 12:58
📁 refs	-	2026-03-21 15:45
📄 remote.py # remote.py # Copyright (C) 2008, 2009 Michael Trier (mtrier@gmail.com) and contributors # # This module is part of GitPython and is released under # the BSD License: http://www.opensource.org/licenses/bsd-license.php # Module implementing a remote object allowing easy access to git remotes import logging import re from git.cmd import handle_process_output, Git from git.compat import defenc, force_text from git.exc import GitCommandError from git.util import ( LazyMixin, IterableObj, IterableList, RemoteProgress, CallableRemoteProgress, ) from git.util import ( join_path, ) from git.config import ( GitConfigParser, SectionConstraint, cp, ) from git.refs import Head, Reference, RemoteReference, SymbolicReference, TagReference # typing------------------------------------------------------- from typing import ( Any, Callable, Dict, Iterator, List, NoReturn, Optional, Sequence, TYPE_CHECKING, Type, Union, cast, overload, ) from git.types import PathLike, Literal, Commit_ish if TYPE_CHECKING: from git.repo.base import Repo from git.objects.submodule.base import UpdateProgress # from git.objects.commit import Commit # from git.objects import Blob, Tree, TagObject flagKeyLiteral = Literal[" ", "!", "+", "-", "", "=", "t", "?"] # def is_flagKeyLiteral(inp: str) -> TypeGuard[flagKeyLiteral]: # return inp in [' ', '!', '+', '-', '=', '', 't', '?'] # ------------------------------------------------------------- log = logging.getLogger("git.remote") log.addHandler(logging.NullHandler()) __all__ = ("RemoteProgress", "PushInfo", "FetchInfo", "Remote") # { Utilities def add_progress( kwargs: Any, git: Git, progress: Union[RemoteProgress, "UpdateProgress", Callable[..., RemoteProgress], None], ) -> Any: """Add the --progress flag to the given kwargs dict if supported by the git command. If the actual progress in the given progress instance is not given, we do not request any progress :return: possibly altered kwargs""" if progress is not None: v = git.version_info[:2] if v >= (1, 7): kwargs["progress"] = True # END handle --progress # END handle progress return kwargs # } END utilities @overload def to_progress_instance(progress: None) -> RemoteProgress: ... @overload def to_progress_instance(progress: Callable[..., Any]) -> CallableRemoteProgress: ... @overload def to_progress_instance(progress: RemoteProgress) -> RemoteProgress: ... def to_progress_instance( progress: Union[Callable[..., Any], RemoteProgress, None] ) -> Union[RemoteProgress, CallableRemoteProgress]: """Given the 'progress' return a suitable object derived from RemoteProgress(). """ # new API only needs progress as a function if callable(progress): return CallableRemoteProgress(progress) # where None is passed create a parser that eats the progress elif progress is None: return RemoteProgress() # assume its the old API with an instance of RemoteProgress. return progress class PushInfo(IterableObj, object): """ Carries information about the result of a push operation of a single head:: info = remote.push()[0] info.flags # bitflags providing more information about the result info.local_ref # Reference pointing to the local reference that was pushed # It is None if the ref was deleted. info.remote_ref_string # path to the remote reference located on the remote side info.remote_ref # Remote Reference on the local side corresponding to # the remote_ref_string. It can be a TagReference as well. info.old_commit # commit at which the remote_ref was standing before we pushed # it to local_ref.commit. Will be None if an error was indicated info.summary # summary line providing human readable english text about the push """ __slots__ = ( "local_ref", "remote_ref_string", "flags", "_old_commit_sha", "_remote", "summary", ) _id_attribute_ = "pushinfo" ( NEW_TAG, NEW_HEAD, NO_MATCH, REJECTED, REMOTE_REJECTED, REMOTE_FAILURE, DELETED, FORCED_UPDATE, FAST_FORWARD, UP_TO_DATE, ERROR, ) = [1 << x for x in range(11)] _flag_map = { "X": NO_MATCH, "-": DELETED, "": 0, "+": FORCED_UPDATE, " ": FAST_FORWARD, "=": UP_TO_DATE, "!": ERROR, } def __init__( self, flags: int, local_ref: Union[SymbolicReference, None], remote_ref_string: str, remote: "Remote", old_commit: Optional[str] = None, summary: str = "", ) -> None: """Initialize a new instance local_ref: HEAD \| Head \| RemoteReference \| TagReference \| Reference \| SymbolicReference \| None""" self.flags = flags self.local_ref = local_ref self.remote_ref_string = remote_ref_string self._remote = remote self._old_commit_sha = old_commit self.summary = summary @property def old_commit(self) -> Union[str, SymbolicReference, Commit_ish, None]: return self._old_commit_sha and self._remote.repo.commit(self._old_commit_sha) or None @property def remote_ref(self) -> Union[RemoteReference, TagReference]: """ :return: Remote Reference or TagReference in the local repository corresponding to the remote_ref_string kept in this instance.""" # translate heads to a local remote, tags stay as they are if self.remote_ref_string.startswith("refs/tags"): return TagReference(self._remote.repo, self.remote_ref_string) elif self.remote_ref_string.startswith("refs/heads"): remote_ref = Reference(self._remote.repo, self.remote_ref_string) return RemoteReference( self._remote.repo, "refs/remotes/%s/%s" % (str(self._remote), remote_ref.name), ) else: raise ValueError("Could not handle remote ref: %r" % self.remote_ref_string) # END @classmethod def _from_line(cls, remote: "Remote", line: str) -> "PushInfo": """Create a new PushInfo instance as parsed from line which is expected to be like refs/heads/master:refs/heads/master 05d2687..1d0568e as bytes""" control_character, from_to, summary = line.split("\t", 3) flags = 0 # control character handling try: flags \|= cls._flag_map[control_character] except KeyError as e: raise ValueError("Control character %r unknown as parsed from line %r" % (control_character, line)) from e # END handle control character # from_to handling from_ref_string, to_ref_string = from_to.split(":") if flags & cls.DELETED: from_ref: Union[SymbolicReference, None] = None else: if from_ref_string == "(delete)": from_ref = None else: from_ref = Reference.from_path(remote.repo, from_ref_string) # commit handling, could be message or commit info old_commit: Optional[str] = None if summary.startswith("["): if "[rejected]" in summary: flags \|= cls.REJECTED elif "[remote rejected]" in summary: flags \|= cls.REMOTE_REJECTED elif "[remote failure]" in summary: flags \|= cls.REMOTE_FAILURE elif "[no match]" in summary: flags \|= cls.ERROR elif "[new tag]" in summary: flags \|= cls.NEW_TAG elif "[new branch]" in summary: flags \|= cls.NEW_HEAD # uptodate encoded in control character else: # fast-forward or forced update - was encoded in control character, # but we parse the old and new commit split_token = "..." if control_character == " ": split_token = ".." old_sha, _new_sha = summary.split(" ")[0].split(split_token) # have to use constructor here as the sha usually is abbreviated old_commit = old_sha # END message handling return PushInfo(flags, from_ref, to_ref_string, remote, old_commit, summary) @classmethod def iter_items(cls, repo: "Repo", args: Any, *kwargs: Any) -> NoReturn: # -> Iterator['PushInfo']: raise NotImplementedError class PushInfoList(IterableList[PushInfo]): """ IterableList of PushInfo objects. """ def __new__(cls) -> "PushInfoList": return cast(PushInfoList, IterableList.__new__(cls, "push_infos")) def __init__(self) -> None: super().__init__("push_infos") self.error: Optional[Exception] = None def raise_if_error(self) -> None: """ Raise an exception if any ref failed to push. """ if self.error: raise self.error class FetchInfo(IterableObj, object): """ Carries information about the results of a fetch operation of a single head:: info = remote.fetch()[0] info.ref # Symbolic Reference or RemoteReference to the changed # remote head or FETCH_HEAD info.flags # additional flags to be & with enumeration members, # i.e. info.flags & info.REJECTED # is 0 if ref is SymbolicReference info.note # additional notes given by git-fetch intended for the user info.old_commit # if info.flags & info.FORCED_UPDATE\|info.FAST_FORWARD, # field is set to the previous location of ref, otherwise None info.remote_ref_path # The path from which we fetched on the remote. It's the remote's version of our info.ref """ __slots__ = ("ref", "old_commit", "flags", "note", "remote_ref_path") _id_attribute_ = "fetchinfo" ( NEW_TAG, NEW_HEAD, HEAD_UPTODATE, TAG_UPDATE, REJECTED, FORCED_UPDATE, FAST_FORWARD, ERROR, ) = [1 << x for x in range(8)] _re_fetch_result = re.compile(r"^\s(.) (\[[\w\s\.$@]+\]\|[\w\.$@]+)\s+(.+) -> ([^\s]+)( \(.\)?$)?") _flag_map: Dict[flagKeyLiteral, int] = { "!": ERROR, "+": FORCED_UPDATE, "": 0, "=": HEAD_UPTODATE, " ": FAST_FORWARD, "-": TAG_UPDATE, } @classmethod def refresh(cls) -> Literal[True]: """This gets called by the refresh function (see the top level __init__). """ # clear the old values in _flag_map try: del cls._flag_map["t"] except KeyError: pass try: del cls._flag_map["-"] except KeyError: pass # set the value given the git version if Git().version_info[:2] >= (2, 10): cls._flag_map["t"] = cls.TAG_UPDATE else: cls._flag_map["-"] = cls.TAG_UPDATE return True def __init__( self, ref: SymbolicReference, flags: int, note: str = "", old_commit: Union[Commit_ish, None] = None, remote_ref_path: Optional[PathLike] = None, ) -> None: """ Initialize a new instance """ self.ref = ref self.flags = flags self.note = note self.old_commit = old_commit self.remote_ref_path = remote_ref_path def __str__(self) -> str: return self.name @property def name(self) -> str: """:return: Name of our remote ref""" return self.ref.name @property def commit(self) -> Commit_ish: """:return: Commit of our remote ref""" return self.ref.commit @classmethod def _from_line(cls, repo: "Repo", line: str, fetch_line: str) -> "FetchInfo": """Parse information from the given line as returned by git-fetch -v and return a new FetchInfo object representing this information. We can handle a line as follows: "%c %-\\s %-\\s -> %s%s" Where c is either ' ', !, +, -, \\, or = ! means error + means success forcing update - means a tag was updated means birth of new branch or tag = means the head was up to date ( and not moved ) ' ' means a fast-forward fetch line is the corresponding line from FETCH_HEAD, like acb0fa8b94ef421ad60c8507b634759a472cd56c not-for-merge branch '0.1.7RC' of /tmp/tmpya0vairemote_repo""" match = cls._re_fetch_result.match(line) if match is None: raise ValueError("Failed to parse line: %r" % line) # parse lines remote_local_ref_str: str ( control_character, operation, local_remote_ref, remote_local_ref_str, note, ) = match.groups() # assert is_flagKeyLiteral(control_character), f"{control_character}" control_character = cast(flagKeyLiteral, control_character) try: _new_hex_sha, _fetch_operation, fetch_note = fetch_line.split("\t") ref_type_name, fetch_note = fetch_note.split(" ", 1) except ValueError as e: # unpack error raise ValueError("Failed to parse FETCH_HEAD line: %r" % fetch_line) from e # parse flags from control_character flags = 0 try: flags \|= cls._flag_map[control_character] except KeyError as e: raise ValueError("Control character %r unknown as parsed from line %r" % (control_character, line)) from e # END control char exception handling # parse operation string for more info - makes no sense for symbolic refs, but we parse it anyway old_commit: Union[Commit_ish, None] = None is_tag_operation = False if "rejected" in operation: flags \|= cls.REJECTED if "new tag" in operation: flags \|= cls.NEW_TAG is_tag_operation = True if "tag update" in operation: flags \|= cls.TAG_UPDATE is_tag_operation = True if "new branch" in operation: flags \|= cls.NEW_HEAD if "..." in operation or ".." in operation: split_token = "..." if control_character == " ": split_token = split_token[:-1] old_commit = repo.rev_parse(operation.split(split_token)[0]) # END handle refspec # handle FETCH_HEAD and figure out ref type # If we do not specify a target branch like master:refs/remotes/origin/master, # the fetch result is stored in FETCH_HEAD which destroys the rule we usually # have. In that case we use a symbolic reference which is detached ref_type: Optional[Type[SymbolicReference]] = None if remote_local_ref_str == "FETCH_HEAD": ref_type = SymbolicReference elif ref_type_name == "tag" or is_tag_operation: # the ref_type_name can be branch, whereas we are still seeing a tag operation. It happens during # testing, which is based on actual git operations ref_type = TagReference elif ref_type_name in ("remote-tracking", "branch"): # note: remote-tracking is just the first part of the 'remote-tracking branch' token. # We don't parse it correctly, but its enough to know what to do, and its new in git 1.7something ref_type = RemoteReference elif "/" in ref_type_name: # If the fetch spec look something like this '+refs/pull/:refs/heads/pull/', and is thus pretty # much anything the user wants, we will have trouble to determine what's going on # For now, we assume the local ref is a Head ref_type = Head else: raise TypeError("Cannot handle reference type: %r" % ref_type_name) # END handle ref type # create ref instance if ref_type is SymbolicReference: remote_local_ref = ref_type(repo, "FETCH_HEAD") else: # determine prefix. Tags are usually pulled into refs/tags, they may have subdirectories. # It is not clear sometimes where exactly the item is, unless we have an absolute path as indicated # by the 'ref/' prefix. Otherwise even a tag could be in refs/remotes, which is when it will have the # 'tags/' subdirectory in its path. # We don't want to test for actual existence, but try to figure everything out analytically. ref_path: Optional[PathLike] = None remote_local_ref_str = remote_local_ref_str.strip() if remote_local_ref_str.startswith(Reference._common_path_default + "/"): # always use actual type if we get absolute paths # Will always be the case if something is fetched outside of refs/remotes (if its not a tag) ref_path = remote_local_ref_str if ref_type is not TagReference and not remote_local_ref_str.startswith( RemoteReference._common_path_default + "/" ): ref_type = Reference # END downgrade remote reference elif ref_type is TagReference and "tags/" in remote_local_ref_str: # even though its a tag, it is located in refs/remotes ref_path = join_path(RemoteReference._common_path_default, remote_local_ref_str) else: ref_path = join_path(ref_type._common_path_default, remote_local_ref_str) # END obtain refpath # even though the path could be within the git conventions, we make # sure we respect whatever the user wanted, and disabled path checking remote_local_ref = ref_type(repo, ref_path, check_path=False) # END create ref instance note = (note and note.strip()) or "" return cls(remote_local_ref, flags, note, old_commit, local_remote_ref) @classmethod def iter_items(cls, repo: "Repo", args: Any, kwargs: Any) -> NoReturn: # -> Iterator['FetchInfo']: raise NotImplementedError class Remote(LazyMixin, IterableObj): """Provides easy read and write access to a git remote. Everything not part of this interface is considered an option for the current remote, allowing constructs like remote.pushurl to query the pushurl. NOTE: When querying configuration, the configuration accessor will be cached to speed up subsequent accesses.""" __slots__ = ("repo", "name", "_config_reader") _id_attribute_ = "name" unsafe_git_fetch_options = [ # This option allows users to execute arbitrary commands. # https://git-scm.com/docs/git-fetch#Documentation/git-fetch.txt---upload-packltupload-packgt "--upload-pack", ] unsafe_git_pull_options = [ # This option allows users to execute arbitrary commands. # https://git-scm.com/docs/git-pull#Documentation/git-pull.txt---upload-packltupload-packgt "--upload-pack" ] unsafe_git_push_options = [ # This option allows users to execute arbitrary commands. # https://git-scm.com/docs/git-push#Documentation/git-push.txt---execltgit-receive-packgt "--receive-pack", "--exec", ] def __init__(self, repo: "Repo", name: str) -> None: """Initialize a remote instance :param repo: The repository we are a remote of :param name: the name of the remote, i.e. 'origin'""" self.repo = repo self.name = name self.url: str def __getattr__(self, attr: str) -> Any: """Allows to call this instance like remote.special( \\args, \\\\kwargs) to call git-remote special self.name""" if attr == "_config_reader": return super(Remote, self).__getattr__(attr) # sometimes, probably due to a bug in python itself, we are being called # even though a slot of the same name exists try: return self._config_reader.get(attr) except cp.NoOptionError: return super(Remote, self).__getattr__(attr) # END handle exception def _config_section_name(self) -> str: return 'remote "%s"' % self.name def _set_cache_(self, attr: str) -> None: if attr == "_config_reader": # NOTE: This is cached as __getattr__ is overridden to return remote config values implicitly, such as # in print(r.pushurl) self._config_reader = SectionConstraint(self.repo.config_reader("repository"), self._config_section_name()) else: super(Remote, self)._set_cache_(attr) def __str__(self) -> str: return self.name def __repr__(self) -> str: return '<git.%s "%s">' % (self.__class__.__name__, self.name) def __eq__(self, other: object) -> bool: return isinstance(other, type(self)) and self.name == other.name def __ne__(self, other: object) -> bool: return not (self == other) def __hash__(self) -> int: return hash(self.name) def exists(self) -> bool: """ :return: True if this is a valid, existing remote. Valid remotes have an entry in the repository's configuration""" try: self.config_reader.get("url") return True except cp.NoOptionError: # we have the section at least ... return True except cp.NoSectionError: return False # end @classmethod def iter_items(cls, repo: "Repo", args: Any, kwargs: Any) -> Iterator["Remote"]: """:return: Iterator yielding Remote objects of the given repository""" for section in repo.config_reader("repository").sections(): if not section.startswith("remote "): continue lbound = section.find('"') rbound = section.rfind('"') if lbound == -1 or rbound == -1: raise ValueError("Remote-Section has invalid format: %r" % section) yield Remote(repo, section[lbound + 1 : rbound]) # END for each configuration section def set_url( self, new_url: str, old_url: Optional[str] = None, allow_unsafe_protocols: bool = False, kwargs: Any ) -> "Remote": """Configure URLs on current remote (cf command git remote set_url) This command manages URLs on the remote. :param new_url: string being the URL to add as an extra remote URL :param old_url: when set, replaces this URL with new_url for the remote :param allow_unsafe_protocols: Allow unsafe protocols to be used, like ext :return: self """ if not allow_unsafe_protocols: Git.check_unsafe_protocols(new_url) scmd = "set-url" kwargs["insert_kwargs_after"] = scmd if old_url: self.repo.git.remote(scmd, "--", self.name, new_url, old_url, kwargs) else: self.repo.git.remote(scmd, "--", self.name, new_url, kwargs) return self def add_url(self, url: str, allow_unsafe_protocols: bool = False, kwargs: Any) -> "Remote": """Adds a new url on current remote (special case of git remote set_url) This command adds new URLs to a given remote, making it possible to have multiple URLs for a single remote. :param url: string being the URL to add as an extra remote URL :param allow_unsafe_protocols: Allow unsafe protocols to be used, like ext :return: self """ return self.set_url(url, add=True, allow_unsafe_protocols=allow_unsafe_protocols) def delete_url(self, url: str, kwargs: Any) -> "Remote": """Deletes a new url on current remote (special case of git remote set_url) This command deletes new URLs to a given remote, making it possible to have multiple URLs for a single remote. :param url: string being the URL to delete from the remote :return: self """ return self.set_url(url, delete=True) @property def urls(self) -> Iterator[str]: """:return: Iterator yielding all configured URL targets on a remote as strings""" try: remote_details = self.repo.git.remote("get-url", "--all", self.name) assert isinstance(remote_details, str) for line in remote_details.split("\n"): yield line except GitCommandError as ex: ## We are on git < 2.7 (i.e TravisCI as of Oct-2016), # so `get-utl` command does not exist yet! # see: https://github.com/gitpython-developers/GitPython/pull/528#issuecomment-252976319 # and: http://stackoverflow.com/a/32991784/548792 # if "Unknown subcommand: get-url" in str(ex): try: remote_details = self.repo.git.remote("show", self.name) assert isinstance(remote_details, str) for line in remote_details.split("\n"): if " Push URL:" in line: yield line.split(": ")[-1] except GitCommandError as _ex: if any(msg in str(_ex) for msg in ["correct access rights", "cannot run ssh"]): # If ssh is not setup to access this repository, see issue 694 remote_details = self.repo.git.config("--get-all", "remote.%s.url" % self.name) assert isinstance(remote_details, str) for line in remote_details.split("\n"): yield line else: raise _ex else: raise ex @property def refs(self) -> IterableList[RemoteReference]: """ :return: IterableList of RemoteReference objects. It is prefixed, allowing you to omit the remote path portion, i.e.:: remote.refs.master # yields RemoteReference('/refs/remotes/origin/master')""" out_refs: IterableList[RemoteReference] = IterableList(RemoteReference._id_attribute_, "%s/" % self.name) out_refs.extend(RemoteReference.list_items(self.repo, remote=self.name)) return out_refs @property def stale_refs(self) -> IterableList[Reference]: """ :return: IterableList RemoteReference objects that do not have a corresponding head in the remote reference anymore as they have been deleted on the remote side, but are still available locally. The IterableList is prefixed, hence the 'origin' must be omitted. See 'refs' property for an example. To make things more complicated, it can be possible for the list to include other kinds of references, for example, tag references, if these are stale as well. This is a fix for the issue described here: https://github.com/gitpython-developers/GitPython/issues/260 """ out_refs: IterableList[Reference] = IterableList(RemoteReference._id_attribute_, "%s/" % self.name) for line in self.repo.git.remote("prune", "--dry-run", self).splitlines()[2:]: # expecting # [would prune] origin/new_branch token = " * [would prune] " if not line.startswith(token): continue ref_name = line.replace(token, "") # sometimes, paths start with a full ref name, like refs/tags/foo, see #260 if ref_name.startswith(Reference._common_path_default + "/"): out_refs.append(Reference.from_path(self.repo, ref_name)) else: fqhn = "%s/%s" % (RemoteReference._common_path_default, ref_name) out_refs.append(RemoteReference(self.repo, fqhn)) # end special case handling # END for each line return out_refs @classmethod def create(cls, repo: "Repo", name: str, url: str, allow_unsafe_protocols: bool = False, kwargs: Any) -> "Remote": """Create a new remote to the given repository :param repo: Repository instance that is to receive the new remote :param name: Desired name of the remote :param url: URL which corresponds to the remote's name :param allow_unsafe_protocols: Allow unsafe protocols to be used, like ext :param kwargs: Additional arguments to be passed to the git-remote add command :return: New Remote instance :raise GitCommandError: in case an origin with that name already exists""" scmd = "add" kwargs["insert_kwargs_after"] = scmd url = Git.polish_url(url) if not allow_unsafe_protocols: Git.check_unsafe_protocols(url) repo.git.remote(scmd, "--", name, url, kwargs) return cls(repo, name) # add is an alias @classmethod def add(cls, repo: "Repo", name: str, url: str, kwargs: Any) -> "Remote": return cls.create(repo, name, url, kwargs) @classmethod def remove(cls, repo: "Repo", name: str) -> str: """Remove the remote with the given name :return: the passed remote name to remove """ repo.git.remote("rm", name) if isinstance(name, cls): name._clear_cache() return name # alias rm = remove def rename(self, new_name: str) -> "Remote": """Rename self to the given new_name :return: self""" if self.name == new_name: return self self.repo.git.remote("rename", self.name, new_name) self.name = new_name self._clear_cache() return self def update(self, kwargs: Any) -> "Remote": """Fetch all changes for this remote, including new branches which will be forced in ( in case your local remote branch is not part the new remote branches ancestry anymore ). :param kwargs: Additional arguments passed to git-remote update :return: self""" scmd = "update" kwargs["insert_kwargs_after"] = scmd self.repo.git.remote(scmd, self.name, kwargs) return self def _get_fetch_info_from_stderr( self, proc: "Git.AutoInterrupt", progress: Union[Callable[..., Any], RemoteProgress, None], kill_after_timeout: Union[None, float] = None, ) -> IterableList["FetchInfo"]: progress = to_progress_instance(progress) # skip first line as it is some remote info we are not interested in output: IterableList["FetchInfo"] = IterableList("name") # lines which are no progress are fetch info lines # this also waits for the command to finish # Skip some progress lines that don't provide relevant information fetch_info_lines = [] # Basically we want all fetch info lines which appear to be in regular form, and thus have a # command character. Everything else we ignore, cmds = set(FetchInfo._flag_map.keys()) progress_handler = progress.new_message_handler() handle_process_output( proc, None, progress_handler, finalizer=None, decode_streams=False, kill_after_timeout=kill_after_timeout, ) stderr_text = progress.error_lines and "\n".join(progress.error_lines) or "" proc.wait(stderr=stderr_text) if stderr_text: log.warning("Error lines received while fetching: %s", stderr_text) for line in progress.other_lines: line = force_text(line) for cmd in cmds: if len(line) > 1 and line[0] == " " and line[1] == cmd: fetch_info_lines.append(line) continue # read head information fetch_head = SymbolicReference(self.repo, "FETCH_HEAD") with open(fetch_head.abspath, "rb") as fp: fetch_head_info = [line.decode(defenc) for line in fp.readlines()] l_fil = len(fetch_info_lines) l_fhi = len(fetch_head_info) if l_fil != l_fhi: msg = "Fetch head lines do not match lines provided via progress information\n" msg += "length of progress lines %i should be equal to lines in FETCH_HEAD file %i\n" msg += "Will ignore extra progress lines or fetch head lines." msg %= (l_fil, l_fhi) log.debug(msg) log.debug(b"info lines: " + str(fetch_info_lines).encode("UTF-8")) log.debug(b"head info: " + str(fetch_head_info).encode("UTF-8")) if l_fil < l_fhi: fetch_head_info = fetch_head_info[:l_fil] else: fetch_info_lines = fetch_info_lines[:l_fhi] # end truncate correct list # end sanity check + sanitization for err_line, fetch_line in zip(fetch_info_lines, fetch_head_info): try: output.append(FetchInfo._from_line(self.repo, err_line, fetch_line)) except ValueError as exc: log.debug("Caught error while parsing line: %s", exc) log.warning("Git informed while fetching: %s", err_line.strip()) return output def _get_push_info( self, proc: "Git.AutoInterrupt", progress: Union[Callable[..., Any], RemoteProgress, None], kill_after_timeout: Union[None, float] = None, ) -> PushInfoList: progress = to_progress_instance(progress) # read progress information from stderr # we hope stdout can hold all the data, it should ... # read the lines manually as it will use carriage returns between the messages # to override the previous one. This is why we read the bytes manually progress_handler = progress.new_message_handler() output: PushInfoList = PushInfoList() def stdout_handler(line: str) -> None: try: output.append(PushInfo._from_line(self, line)) except ValueError: # If an error happens, additional info is given which we parse below. pass handle_process_output( proc, stdout_handler, progress_handler, finalizer=None, decode_streams=False, kill_after_timeout=kill_after_timeout, ) stderr_text = progress.error_lines and "\n".join(progress.error_lines) or "" try: proc.wait(stderr=stderr_text) except Exception as e: # This is different than fetch (which fails if there is any std_err # even if there is an output) if not output: raise elif stderr_text: log.warning("Error lines received while fetching: %s", stderr_text) output.error = e return output def _assert_refspec(self) -> None: """Turns out we can't deal with remotes if the refspec is missing""" config = self.config_reader unset = "placeholder" try: if config.get_value("fetch", default=unset) is unset: msg = "Remote '%s' has no refspec set.\n" msg += "You can set it as follows:" msg += " 'git config --add \"remote.%s.fetch +refs/heads/:refs/heads/\"'." raise AssertionError(msg % (self.name, self.name)) finally: config.release() def fetch( self, refspec: Union[str, List[str], None] = None, progress: Union[RemoteProgress, None, "UpdateProgress"] = None, verbose: bool = True, kill_after_timeout: Union[None, float] = None, allow_unsafe_protocols: bool = False, allow_unsafe_options: bool = False, *kwargs: Any, ) -> IterableList[FetchInfo]: """Fetch the latest changes for this remote :param refspec: A "refspec" is used by fetch and push to describe the mapping between remote ref and local ref. They are combined with a colon in the format <src>:<dst>, preceded by an optional plus sign, +. For example: git fetch $URL refs/heads/master:refs/heads/origin means "grab the master branch head from the $URL and store it as my origin branch head". And git push $URL refs/heads/master:refs/heads/to-upstream means "publish my master branch head as to-upstream branch at $URL". See also git-push(1). Taken from the git manual Fetch supports multiple refspecs (as the underlying git-fetch does) - supplying a list rather than a string for 'refspec' will make use of this facility. :param progress: See 'push' method :param verbose: Boolean for verbose output :param kill_after_timeout: To specify a timeout in seconds for the git command, after which the process should be killed. It is set to None by default. :param allow_unsafe_protocols: Allow unsafe protocols to be used, like ext :param allow_unsafe_options: Allow unsafe options to be used, like --upload-pack :param kwargs: Additional arguments to be passed to git-fetch :return: IterableList(FetchInfo, ...) list of FetchInfo instances providing detailed information about the fetch results :note: As fetch does not provide progress information to non-ttys, we cannot make it available here unfortunately as in the 'push' method.""" if refspec is None: # No argument refspec, then ensure the repo's config has a fetch refspec. self._assert_refspec() kwargs = add_progress(kwargs, self.repo.git, progress) if isinstance(refspec, list): args: Sequence[Optional[str]] = refspec else: args = [refspec] if not allow_unsafe_protocols: for ref in args: if ref: Git.check_unsafe_protocols(ref) if not allow_unsafe_options: Git.check_unsafe_options(options=list(kwargs.keys()), unsafe_options=self.unsafe_git_fetch_options) proc = self.repo.git.fetch( "--", self, args, as_process=True, with_stdout=False, universal_newlines=True, v=verbose, kwargs ) res = self._get_fetch_info_from_stderr(proc, progress, kill_after_timeout=kill_after_timeout) if hasattr(self.repo.odb, "update_cache"): self.repo.odb.update_cache() return res def pull( self, refspec: Union[str, List[str], None] = None, progress: Union[RemoteProgress, "UpdateProgress", None] = None, kill_after_timeout: Union[None, float] = None, allow_unsafe_protocols: bool = False, allow_unsafe_options: bool = False, kwargs: Any, ) -> IterableList[FetchInfo]: """Pull changes from the given branch, being the same as a fetch followed by a merge of branch with your local branch. :param refspec: see :meth:`fetch` method :param progress: see :meth:`push` method :param kill_after_timeout: see :meth:`fetch` method :param allow_unsafe_protocols: Allow unsafe protocols to be used, like ext :param allow_unsafe_options: Allow unsafe options to be used, like --upload-pack :param kwargs: Additional arguments to be passed to git-pull :return: Please see :meth:`fetch` method""" if refspec is None: # No argument refspec, then ensure the repo's config has a fetch refspec. self._assert_refspec() kwargs = add_progress(kwargs, self.repo.git, progress) refspec = Git._unpack_args(refspec or []) if not allow_unsafe_protocols: for ref in refspec: Git.check_unsafe_protocols(ref) if not allow_unsafe_options: Git.check_unsafe_options(options=list(kwargs.keys()), unsafe_options=self.unsafe_git_pull_options) proc = self.repo.git.pull( "--", self, refspec, with_stdout=False, as_process=True, universal_newlines=True, v=True, kwargs ) res = self._get_fetch_info_from_stderr(proc, progress, kill_after_timeout=kill_after_timeout) if hasattr(self.repo.odb, "update_cache"): self.repo.odb.update_cache() return res def push( self, refspec: Union[str, List[str], None] = None, progress: Union[RemoteProgress, "UpdateProgress", Callable[..., RemoteProgress], None] = None, kill_after_timeout: Union[None, float] = None, allow_unsafe_protocols: bool = False, allow_unsafe_options: bool = False, kwargs: Any, ) -> PushInfoList: """Push changes from source branch in refspec to target branch in refspec. :param refspec: see 'fetch' method :param progress: Can take one of many value types: * None to discard progress information * A function (callable) that is called with the progress information. Signature: ``progress(op_code, cur_count, max_count=None, message='')``. `Click here <http://goo.gl/NPa7st>`__ for a description of all arguments given to the function. * An instance of a class derived from ``git.RemoteProgress`` that overrides the ``update()`` function. :note: No further progress information is returned after push returns. :param kill_after_timeout: To specify a timeout in seconds for the git command, after which the process should be killed. It is set to None by default. :param allow_unsafe_protocols: Allow unsafe protocols to be used, like ext :param allow_unsafe_options: Allow unsafe options to be used, like --receive-pack :param kwargs: Additional arguments to be passed to git-push :return: A ``PushInfoList`` object, where each list member represents an individual head which had been updated on the remote side. If the push contains rejected heads, these will have the PushInfo.ERROR bit set in their flags. If the operation fails completely, the length of the returned PushInfoList will be 0. Call ``.raise_if_error()`` on the returned object to raise on any failure.""" kwargs = add_progress(kwargs, self.repo.git, progress) refspec = Git._unpack_args(refspec or []) if not allow_unsafe_protocols: for ref in refspec: Git.check_unsafe_protocols(ref) if not allow_unsafe_options: Git.check_unsafe_options(options=list(kwargs.keys()), unsafe_options=self.unsafe_git_push_options) proc = self.repo.git.push( "--", self, refspec, porcelain=True, as_process=True, universal_newlines=True, kill_after_timeout=kill_after_timeout, **kwargs, ) return self._get_push_info(proc, progress, kill_after_timeout=kill_after_timeout) @property def config_reader(self) -> SectionConstraint[GitConfigParser]: """ :return: GitConfigParser compatible object able to read options for only our remote. Hence you may simple type config.get("pushurl") to obtain the information""" return self._config_reader def _clear_cache(self) -> None: try: del self._config_reader except AttributeError: pass # END handle exception @property def config_writer(self) -> SectionConstraint: """ :return: GitConfigParser compatible object able to write options for this remote. :note: You can only own one writer at a time - delete it to release the configuration file and make it usable by others. To assure consistent results, you should only query options through the writer. Once you are done writing, you are free to use the config reader once again.""" writer = self.repo.config_writer() # clear our cache to assure we re-read the possibly changed configuration self._clear_cache() return SectionConstraint(writer, self._config_section_name())	44.01 KB	2026-01-20 12:58
📁 repo	-	2026-03-21 15:45
📄 types.py # -- coding: utf-8 -- # This module is part of GitPython and is released under # the BSD License: http://www.opensource.org/licenses/bsd-license.php # flake8: noqa import os import sys from typing import ( Dict, NoReturn, Sequence, Tuple, Union, Any, TYPE_CHECKING, TypeVar, ) # noqa: F401 if sys.version_info[:2] >= (3, 8): from typing import ( Literal, SupportsIndex, TypedDict, Protocol, runtime_checkable, ) # noqa: F401 else: from typing_extensions import ( Literal, SupportsIndex, # noqa: F401 TypedDict, Protocol, runtime_checkable, ) # noqa: F401 # if sys.version_info[:2] >= (3, 10): # from typing import TypeGuard # noqa: F401 # else: # from typing_extensions import TypeGuard # noqa: F401 if sys.version_info[:2] < (3, 9): PathLike = Union[str, os.PathLike] else: # os.PathLike only becomes subscriptable from Python 3.9 onwards PathLike = Union[str, os.PathLike[str]] if TYPE_CHECKING: from git.repo import Repo from git.objects import Commit, Tree, TagObject, Blob # from git.refs import SymbolicReference TBD = Any _T = TypeVar("_T") Tree_ish = Union["Commit", "Tree"] Commit_ish = Union["Commit", "TagObject", "Blob", "Tree"] Lit_commit_ish = Literal["commit", "tag", "blob", "tree"] # Config_levels --------------------------------------------------------- Lit_config_levels = Literal["system", "global", "user", "repository"] # def is_config_level(inp: str) -> TypeGuard[Lit_config_levels]: # # return inp in get_args(Lit_config_level) # only py >= 3.8 # return inp in ("system", "user", "global", "repository") ConfigLevels_Tup = Tuple[Literal["system"], Literal["user"], Literal["global"], Literal["repository"]] # ----------------------------------------------------------------------------------- def assert_never(inp: NoReturn, raise_error: bool = True, exc: Union[Exception, None] = None) -> None: """For use in exhaustive checking of literal or Enum in if/else chain. Should only be reached if all members not handled OR attempt to pass non-members through chain. If all members handled, type is Empty. Otherwise, will cause mypy error. If non-members given, should cause mypy error at variable creation. If raise_error is True, will also raise AssertionError or the Exception passed to exc. """ if raise_error: if exc is None: raise ValueError(f"An unhandled Literal ({inp}) in an if/else chain was found") else: raise exc class Files_TD(TypedDict): insertions: int deletions: int lines: int class Total_TD(TypedDict): insertions: int deletions: int lines: int files: int class HSH_TD(TypedDict): total: Total_TD files: Dict[PathLike, Files_TD] @runtime_checkable class Has_Repo(Protocol): repo: "Repo" @runtime_checkable class Has_id_attribute(Protocol): _id_attribute_: str	2.95 KB	2026-01-20 12:58
📄 util.py # utils.py # Copyright (C) 2008, 2009 Michael Trier (mtrier@gmail.com) and contributors # # This module is part of GitPython and is released under # the BSD License: http://www.opensource.org/licenses/bsd-license.php from abc import abstractmethod import os.path as osp from .compat import is_win import contextlib from functools import wraps import getpass import logging import os import platform import subprocess import re import shutil import stat from sys import maxsize import time from urllib.parse import urlsplit, urlunsplit import warnings # from git.objects.util import Traversable # typing --------------------------------------------------------- from typing import ( Any, AnyStr, BinaryIO, Callable, Dict, Generator, IO, Iterator, List, Optional, Pattern, Sequence, Tuple, TypeVar, Union, cast, TYPE_CHECKING, overload, ) import pathlib if TYPE_CHECKING: from git.remote import Remote from git.repo.base import Repo from git.config import GitConfigParser, SectionConstraint from git import Git # from git.objects.base import IndexObject from .types import ( Literal, SupportsIndex, Protocol, runtime_checkable, # because behind py version guards PathLike, HSH_TD, Total_TD, Files_TD, # aliases Has_id_attribute, ) T_IterableObj = TypeVar("T_IterableObj", bound=Union["IterableObj", "Has_id_attribute"], covariant=True) # So IterableList[Head] is subtype of IterableList[IterableObj] # --------------------------------------------------------------------- from gitdb.util import ( # NOQA @IgnorePep8 make_sha, LockedFD, # @UnusedImport file_contents_ro, # @UnusedImport file_contents_ro_filepath, # @UnusedImport LazyMixin, # @UnusedImport to_hex_sha, # @UnusedImport to_bin_sha, # @UnusedImport bin_to_hex, # @UnusedImport hex_to_bin, # @UnusedImport ) # NOTE: Some of the unused imports might be used/imported by others. # Handle once test-cases are back up and running. # Most of these are unused here, but are for use by git-python modules so these # don't see gitdb all the time. Flake of course doesn't like it. __all__ = [ "stream_copy", "join_path", "to_native_path_linux", "join_path_native", "Stats", "IndexFileSHA1Writer", "IterableObj", "IterableList", "BlockingLockFile", "LockFile", "Actor", "get_user_id", "assure_directory_exists", "RemoteProgress", "CallableRemoteProgress", "rmtree", "unbare_repo", "HIDE_WINDOWS_KNOWN_ERRORS", ] log = logging.getLogger(__name__) # types############################################################ #: We need an easy way to see if Appveyor TCs start failing, #: so the errors marked with this var are considered "acknowledged" ones, awaiting remedy, #: till then, we wish to hide them. HIDE_WINDOWS_KNOWN_ERRORS = is_win and os.environ.get("HIDE_WINDOWS_KNOWN_ERRORS", True) HIDE_WINDOWS_FREEZE_ERRORS = is_win and os.environ.get("HIDE_WINDOWS_FREEZE_ERRORS", True) # { Utility Methods T = TypeVar("T") def unbare_repo(func: Callable[..., T]) -> Callable[..., T]: """Methods with this decorator raise :class:`.exc.InvalidGitRepositoryError` if they encounter a bare repository""" from .exc import InvalidGitRepositoryError @wraps(func) def wrapper(self: "Remote", args: Any, kwargs: Any) -> T: if self.repo.bare: raise InvalidGitRepositoryError("Method '%s' cannot operate on bare repositories" % func.__name__) # END bare method return func(self, args, *kwargs) # END wrapper return wrapper @contextlib.contextmanager def cwd(new_dir: PathLike) -> Generator[PathLike, None, None]: old_dir = os.getcwd() os.chdir(new_dir) try: yield new_dir finally: os.chdir(old_dir) def rmtree(path: PathLike) -> None: """Remove the given recursively. :note: we use shutil rmtree but adjust its behaviour to see whether files that couldn't be deleted are read-only. Windows will not remove them in that case""" def onerror(func: Callable, path: PathLike, exc_info: str) -> None: # Is the error an access error ? os.chmod(path, stat.S_IWUSR) try: func(path) # Will scream if still not possible to delete. except Exception as ex: if HIDE_WINDOWS_KNOWN_ERRORS: from unittest import SkipTest raise SkipTest("FIXME: fails with: PermissionError\n {}".format(ex)) from ex raise return shutil.rmtree(path, False, onerror) def rmfile(path: PathLike) -> None: """Ensure file deleted also on Windows* where read-only files need special treatment.""" if osp.isfile(path): if is_win: os.chmod(path, 0o777) os.remove(path) def stream_copy(source: BinaryIO, destination: BinaryIO, chunk_size: int = 512 * 1024) -> int: """Copy all data from the source stream into the destination stream in chunks of size chunk_size :return: amount of bytes written""" br = 0 while True: chunk = source.read(chunk_size) destination.write(chunk) br += len(chunk) if len(chunk) < chunk_size: break # END reading output stream return br def join_path(a: PathLike, p: PathLike) -> PathLike: """Join path tokens together similar to osp.join, but always use '/' instead of possibly '\' on windows.""" path = str(a) for b in p: b = str(b) if not b: continue if b.startswith("/"): path += b[1:] elif path == "" or path.endswith("/"): path += b else: path += "/" + b # END for each path token to add return path if is_win: def to_native_path_windows(path: PathLike) -> PathLike: path = str(path) return path.replace("/", "\\") def to_native_path_linux(path: PathLike) -> str: path = str(path) return path.replace("\\", "/") __all__.append("to_native_path_windows") to_native_path = to_native_path_windows else: # no need for any work on linux def to_native_path_linux(path: PathLike) -> str: return str(path) to_native_path = to_native_path_linux def join_path_native(a: PathLike, p: PathLike) -> PathLike: """ As join path, but makes sure an OS native path is returned. This is only needed to play it safe on my dear windows and to assure nice paths that only use '\'""" return to_native_path(join_path(a, p)) def assure_directory_exists(path: PathLike, is_file: bool = False) -> bool: """Assure that the directory pointed to by path exists. :param is_file: If True, path is assumed to be a file and handled correctly. Otherwise it must be a directory :return: True if the directory was created, False if it already existed""" if is_file: path = osp.dirname(path) # END handle file if not osp.isdir(path): os.makedirs(path, exist_ok=True) return True return False def _get_exe_extensions() -> Sequence[str]: PATHEXT = os.environ.get("PATHEXT", None) return ( tuple(p.upper() for p in PATHEXT.split(os.pathsep)) if PATHEXT else (".BAT", "COM", ".EXE") if is_win else ("") ) def py_where(program: str, path: Optional[PathLike] = None) -> List[str]: # From: http://stackoverflow.com/a/377028/548792 winprog_exts = _get_exe_extensions() def is_exec(fpath: str) -> bool: return ( osp.isfile(fpath) and os.access(fpath, os.X_OK) and (os.name != "nt" or not winprog_exts or any(fpath.upper().endswith(ext) for ext in winprog_exts)) ) progs = [] if not path: path = os.environ["PATH"] for folder in str(path).split(os.pathsep): folder = folder.strip('"') if folder: exe_path = osp.join(folder, program) for f in [exe_path] + ["%s%s" % (exe_path, e) for e in winprog_exts]: if is_exec(f): progs.append(f) return progs def _cygexpath(drive: Optional[str], path: str) -> str: if osp.isabs(path) and not drive: # Invoked from `cygpath()` directly with `D:Apps\123`? # It's an error, leave it alone just slashes) p = path # convert to str if AnyPath given else: p = path and osp.normpath(osp.expandvars(osp.expanduser(path))) if osp.isabs(p): if drive: # Confusing, maybe a remote system should expand vars. p = path else: p = cygpath(p) elif drive: p = "/proc/cygdrive/%s/%s" % (drive.lower(), p) p_str = str(p) # ensure it is a str and not AnyPath return p_str.replace("\\", "/") _cygpath_parsers: Tuple[Tuple[Pattern[str], Callable, bool], ...] = ( # See: https://msdn.microsoft.com/en-us/library/windows/desktop/aa365247(v=vs.85).aspx # and: https://www.cygwin.com/cygwin-ug-net/using.html#unc-paths ( re.compile(r"\\\\\?\\UNC\\([^\\]+)\\([^\\]+)(?:\\(.))?"), (lambda server, share, rest_path: "//%s/%s/%s" % (server, share, rest_path.replace("\\", "/"))), False, ), (re.compile(r"\\\\\?\\(\w):[/\\](.)"), (_cygexpath), False), (re.compile(r"(\w):[/\\](.)"), (_cygexpath), False), (re.compile(r"file:(.)", re.I), (lambda rest_path: rest_path), True), (re.compile(r"(\w{2,}:.)"), (lambda url: url), False), # remote URL, do nothing ) def cygpath(path: str) -> str: """Use :meth:`git.cmd.Git.polish_url()` instead, that works on any environment.""" path = str(path) # ensure is str and not AnyPath. # Fix to use Paths when 3.5 dropped. or to be just str if only for urls? if not path.startswith(("/cygdrive", "//", "/proc/cygdrive")): for regex, parser, recurse in _cygpath_parsers: match = regex.match(path) if match: path = parser(match.groups()) if recurse: path = cygpath(path) break else: path = _cygexpath(None, path) return path _decygpath_regex = re.compile(r"(?:/proc)?/cygdrive/(\w)(/.)?") def decygpath(path: PathLike) -> str: path = str(path) m = _decygpath_regex.match(path) if m: drive, rest_path = m.groups() path = "%s:%s" % (drive.upper(), rest_path or "") return path.replace("/", "\\") #: Store boolean flags denoting if a specific Git executable #: is from a Cygwin installation (since `cache_lru()` unsupported on PY2). _is_cygwin_cache: Dict[str, Optional[bool]] = {} @overload def is_cygwin_git(git_executable: None) -> Literal[False]: ... @overload def is_cygwin_git(git_executable: PathLike) -> bool: ... def is_cygwin_git(git_executable: Union[None, PathLike]) -> bool: if is_win: # is_win seems to be true only for Windows-native pythons # cygwin has os.name = posix, I think return False if git_executable is None: return False git_executable = str(git_executable) is_cygwin = _is_cygwin_cache.get(git_executable) # type: Optional[bool] if is_cygwin is None: is_cygwin = False try: git_dir = osp.dirname(git_executable) if not git_dir: res = py_where(git_executable) git_dir = osp.dirname(res[0]) if res else "" # Just a name given, not a real path. uname_cmd = osp.join(git_dir, "uname") process = subprocess.Popen([uname_cmd], stdout=subprocess.PIPE, universal_newlines=True) uname_out, _ = process.communicate() # retcode = process.poll() is_cygwin = "CYGWIN" in uname_out except Exception as ex: log.debug("Failed checking if running in CYGWIN due to: %r", ex) _is_cygwin_cache[git_executable] = is_cygwin return is_cygwin def get_user_id() -> str: """:return: string identifying the currently active system user as name@node""" return "%s@%s" % (getpass.getuser(), platform.node()) def finalize_process(proc: Union[subprocess.Popen, "Git.AutoInterrupt"], kwargs: Any) -> None: """Wait for the process (clone, fetch, pull or push) and handle its errors accordingly""" # TODO: No close proc-streams?? proc.wait(kwargs) @overload def expand_path(p: None, expand_vars: bool = ...) -> None: ... @overload def expand_path(p: PathLike, expand_vars: bool = ...) -> str: # improve these overloads when 3.5 dropped ... def expand_path(p: Union[None, PathLike], expand_vars: bool = True) -> Optional[PathLike]: if isinstance(p, pathlib.Path): return p.resolve() try: p = osp.expanduser(p) # type: ignore if expand_vars: p = osp.expandvars(p) # type: ignore return osp.normpath(osp.abspath(p)) # type: ignore except Exception: return None def remove_password_if_present(cmdline: Sequence[str]) -> List[str]: """ Parse any command line argument and if on of the element is an URL with a username and/or password, replace them by stars (in-place). If nothing found just returns the command line as-is. This should be used for every log line that print a command line, as well as exception messages. """ new_cmdline = [] for index, to_parse in enumerate(cmdline): new_cmdline.append(to_parse) try: url = urlsplit(to_parse) # Remove password from the URL if present if url.password is None and url.username is None: continue if url.password is not None: url = url._replace(netloc=url.netloc.replace(url.password, "***")) if url.username is not None: url = url._replace(netloc=url.netloc.replace(url.username, "**")) new_cmdline[index] = urlunsplit(url) except ValueError: # This is not a valid URL continue return new_cmdline # } END utilities # { Classes class RemoteProgress(object): """ Handler providing an interface to parse progress information emitted by git-push and git-fetch and to dispatch callbacks allowing subclasses to react to the progress. """ _num_op_codes: int = 9 ( BEGIN, END, COUNTING, COMPRESSING, WRITING, RECEIVING, RESOLVING, FINDING_SOURCES, CHECKING_OUT, ) = [1 << x for x in range(_num_op_codes)] STAGE_MASK = BEGIN \| END OP_MASK = ~STAGE_MASK DONE_TOKEN = "done." TOKEN_SEPARATOR = ", " __slots__ = ( "_cur_line", "_seen_ops", "error_lines", # Lines that started with 'error:' or 'fatal:'. "other_lines", ) # Lines not denoting progress (i.e.g. push-infos). re_op_absolute = re.compile(r"(remote: )?([\w\s]+):\s+()(\d+)()(.)") re_op_relative = re.compile(r"(remote: )?([\w\s]+):\s+(\d+)% \((\d+)/(\d+)\)(.)") def __init__(self) -> None: self._seen_ops: List[int] = [] self._cur_line: Optional[str] = None self.error_lines: List[str] = [] self.other_lines: List[str] = [] def _parse_progress_line(self, line: AnyStr) -> None: """Parse progress information from the given line as retrieved by git-push or git-fetch. - Lines that do not contain progress info are stored in :attr:`other_lines`. - Lines that seem to contain an error (i.e. start with error: or fatal:) are stored in :attr:`error_lines`.""" # handle # Counting objects: 4, done. # Compressing objects: 50% (1/2) # Compressing objects: 100% (2/2) # Compressing objects: 100% (2/2), done. if isinstance(line, bytes): # mypy argues about ternary assignment line_str = line.decode("utf-8") else: line_str = line self._cur_line = line_str if self._cur_line.startswith(("error:", "fatal:")): self.error_lines.append(self._cur_line) return # find escape characters and cut them away - regex will not work with # them as they are non-ascii. As git might expect a tty, it will send them last_valid_index = None for i, c in enumerate(reversed(line_str)): if ord(c) < 32: # its a slice index last_valid_index = -i - 1 # END character was non-ascii # END for each character in line if last_valid_index is not None: line_str = line_str[:last_valid_index] # END cut away invalid part line_str = line_str.rstrip() cur_count, max_count = None, None match = self.re_op_relative.match(line_str) if match is None: match = self.re_op_absolute.match(line_str) if not match: self.line_dropped(line_str) self.other_lines.append(line_str) return # END could not get match op_code = 0 _remote, op_name, _percent, cur_count, max_count, message = match.groups() # get operation id if op_name == "Counting objects": op_code \|= self.COUNTING elif op_name == "Compressing objects": op_code \|= self.COMPRESSING elif op_name == "Writing objects": op_code \|= self.WRITING elif op_name == "Receiving objects": op_code \|= self.RECEIVING elif op_name == "Resolving deltas": op_code \|= self.RESOLVING elif op_name == "Finding sources": op_code \|= self.FINDING_SOURCES elif op_name == "Checking out files": op_code \|= self.CHECKING_OUT else: # Note: On windows it can happen that partial lines are sent # Hence we get something like "CompreReceiving objects", which is # a blend of "Compressing objects" and "Receiving objects". # This can't really be prevented, so we drop the line verbosely # to make sure we get informed in case the process spits out new # commands at some point. self.line_dropped(line_str) # Note: Don't add this line to the other lines, as we have to silently # drop it return None # END handle op code # figure out stage if op_code not in self._seen_ops: self._seen_ops.append(op_code) op_code \|= self.BEGIN # END begin opcode if message is None: message = "" # END message handling message = message.strip() if message.endswith(self.DONE_TOKEN): op_code \|= self.END message = message[: -len(self.DONE_TOKEN)] # END end message handling message = message.strip(self.TOKEN_SEPARATOR) self.update( op_code, cur_count and float(cur_count), max_count and float(max_count), message, ) def new_message_handler(self) -> Callable[[str], None]: """ :return: a progress handler suitable for handle_process_output(), passing lines on to this Progress handler in a suitable format""" def handler(line: AnyStr) -> None: return self._parse_progress_line(line.rstrip()) # end return handler def line_dropped(self, line: str) -> None: """Called whenever a line could not be understood and was therefore dropped.""" pass def update( self, op_code: int, cur_count: Union[str, float], max_count: Union[str, float, None] = None, message: str = "", ) -> None: """Called whenever the progress changes :param op_code: Integer allowing to be compared against Operation IDs and stage IDs. Stage IDs are BEGIN and END. BEGIN will only be set once for each Operation ID as well as END. It may be that BEGIN and END are set at once in case only one progress message was emitted due to the speed of the operation. Between BEGIN and END, none of these flags will be set Operation IDs are all held within the OP_MASK. Only one Operation ID will be active per call. :param cur_count: Current absolute count of items :param max_count: The maximum count of items we expect. It may be None in case there is no maximum number of items or if it is (yet) unknown. :param message: In case of the 'WRITING' operation, it contains the amount of bytes transferred. It may possibly be used for other purposes as well. You may read the contents of the current line in self._cur_line""" pass class CallableRemoteProgress(RemoteProgress): """An implementation forwarding updates to any callable""" __slots__ = "_callable" def __init__(self, fn: Callable) -> None: self._callable = fn super(CallableRemoteProgress, self).__init__() def update(self, args: Any, *kwargs: Any) -> None: self._callable(args, *kwargs) class Actor(object): """Actors hold information about a person acting on the repository. They can be committers and authors or anything with a name and an email as mentioned in the git log entries.""" # PRECOMPILED REGEX name_only_regex = re.compile(r"<(.)>") name_email_regex = re.compile(r"(.) <(.?)>") # ENVIRONMENT VARIABLES # read when creating new commits env_author_name = "GIT_AUTHOR_NAME" env_author_email = "GIT_AUTHOR_EMAIL" env_committer_name = "GIT_COMMITTER_NAME" env_committer_email = "GIT_COMMITTER_EMAIL" # CONFIGURATION KEYS conf_name = "name" conf_email = "email" __slots__ = ("name", "email") def __init__(self, name: Optional[str], email: Optional[str]) -> None: self.name = name self.email = email def __eq__(self, other: Any) -> bool: return self.name == other.name and self.email == other.email def __ne__(self, other: Any) -> bool: return not (self == other) def __hash__(self) -> int: return hash((self.name, self.email)) def __str__(self) -> str: return self.name if self.name else "" def __repr__(self) -> str: return '<git.Actor "%s <%s>">' % (self.name, self.email) @classmethod def _from_string(cls, string: str) -> "Actor": """Create an Actor from a string. :param string: is the string, which is expected to be in regular git format John Doe <jdoe@example.com> :return: Actor""" m = cls.name_email_regex.search(string) if m: name, email = m.groups() return Actor(name, email) else: m = cls.name_only_regex.search(string) if m: return Actor(m.group(1), None) # assume best and use the whole string as name return Actor(string, None) # END special case name # END handle name/email matching @classmethod def _main_actor( cls, env_name: str, env_email: str, config_reader: Union[None, "GitConfigParser", "SectionConstraint"] = None, ) -> "Actor": actor = Actor("", "") user_id = None # We use this to avoid multiple calls to getpass.getuser() def default_email() -> str: nonlocal user_id if not user_id: user_id = get_user_id() return user_id def default_name() -> str: return default_email().split("@")[0] for attr, evar, cvar, default in ( ("name", env_name, cls.conf_name, default_name), ("email", env_email, cls.conf_email, default_email), ): try: val = os.environ[evar] setattr(actor, attr, val) except KeyError: if config_reader is not None: try: val = config_reader.get("user", cvar) except Exception: val = default() setattr(actor, attr, val) # END config-reader handling if not getattr(actor, attr): setattr(actor, attr, default()) # END handle name # END for each item to retrieve return actor @classmethod def committer(cls, config_reader: Union[None, "GitConfigParser", "SectionConstraint"] = None) -> "Actor": """ :return: Actor instance corresponding to the configured committer. It behaves similar to the git implementation, such that the environment will override configuration values of config_reader. If no value is set at all, it will be generated :param config_reader: ConfigReader to use to retrieve the values from in case they are not set in the environment""" return cls._main_actor(cls.env_committer_name, cls.env_committer_email, config_reader) @classmethod def author(cls, config_reader: Union[None, "GitConfigParser", "SectionConstraint"] = None) -> "Actor": """Same as committer(), but defines the main author. It may be specified in the environment, but defaults to the committer""" return cls._main_actor(cls.env_author_name, cls.env_author_email, config_reader) class Stats(object): """ Represents stat information as presented by git at the end of a merge. It is created from the output of a diff operation. ``Example``:: c = Commit( sha1 ) s = c.stats s.total # full-stat-dict s.files # dict( filepath : stat-dict ) ``stat-dict`` A dictionary with the following keys and values:: deletions = number of deleted lines as int insertions = number of inserted lines as int lines = total number of lines changed as int, or deletions + insertions ``full-stat-dict`` In addition to the items in the stat-dict, it features additional information:: files = number of changed files as int""" __slots__ = ("total", "files") def __init__(self, total: Total_TD, files: Dict[PathLike, Files_TD]): self.total = total self.files = files @classmethod def _list_from_string(cls, repo: "Repo", text: str) -> "Stats": """Create a Stat object from output retrieved by git-diff. :return: git.Stat""" hsh: HSH_TD = { "total": {"insertions": 0, "deletions": 0, "lines": 0, "files": 0}, "files": {}, } for line in text.splitlines(): (raw_insertions, raw_deletions, filename) = line.split("\t") insertions = raw_insertions != "-" and int(raw_insertions) or 0 deletions = raw_deletions != "-" and int(raw_deletions) or 0 hsh["total"]["insertions"] += insertions hsh["total"]["deletions"] += deletions hsh["total"]["lines"] += insertions + deletions hsh["total"]["files"] += 1 files_dict: Files_TD = { "insertions": insertions, "deletions": deletions, "lines": insertions + deletions, } hsh["files"][filename.strip()] = files_dict return Stats(hsh["total"], hsh["files"]) class IndexFileSHA1Writer(object): """Wrapper around a file-like object that remembers the SHA1 of the data written to it. It will write a sha when the stream is closed or if the asked for explicitly using write_sha. Only useful to the indexfile :note: Based on the dulwich project""" __slots__ = ("f", "sha1") def __init__(self, f: IO) -> None: self.f = f self.sha1 = make_sha(b"") def write(self, data: AnyStr) -> int: self.sha1.update(data) return self.f.write(data) def write_sha(self) -> bytes: sha = self.sha1.digest() self.f.write(sha) return sha def close(self) -> bytes: sha = self.write_sha() self.f.close() return sha def tell(self) -> int: return self.f.tell() class LockFile(object): """Provides methods to obtain, check for, and release a file based lock which should be used to handle concurrent access to the same file. As we are a utility class to be derived from, we only use protected methods. Locks will automatically be released on destruction""" __slots__ = ("_file_path", "_owns_lock") def __init__(self, file_path: PathLike) -> None: self._file_path = file_path self._owns_lock = False def __del__(self) -> None: self._release_lock() def _lock_file_path(self) -> str: """:return: Path to lockfile""" return "%s.lock" % (self._file_path) def _has_lock(self) -> bool: """:return: True if we have a lock and if the lockfile still exists :raise AssertionError: if our lock-file does not exist""" return self._owns_lock def _obtain_lock_or_raise(self) -> None: """Create a lock file as flag for other instances, mark our instance as lock-holder :raise IOError: if a lock was already present or a lock file could not be written""" if self._has_lock(): return lock_file = self._lock_file_path() if osp.isfile(lock_file): raise IOError( "Lock for file %r did already exist, delete %r in case the lock is illegal" % (self._file_path, lock_file) ) try: flags = os.O_WRONLY \| os.O_CREAT \| os.O_EXCL if is_win: flags \|= os.O_SHORT_LIVED fd = os.open(lock_file, flags, 0) os.close(fd) except OSError as e: raise IOError(str(e)) from e self._owns_lock = True def _obtain_lock(self) -> None: """The default implementation will raise if a lock cannot be obtained. Subclasses may override this method to provide a different implementation""" return self._obtain_lock_or_raise() def _release_lock(self) -> None: """Release our lock if we have one""" if not self._has_lock(): return # if someone removed our file beforhand, lets just flag this issue # instead of failing, to make it more usable. lfp = self._lock_file_path() try: rmfile(lfp) except OSError: pass self._owns_lock = False class BlockingLockFile(LockFile): """The lock file will block until a lock could be obtained, or fail after a specified timeout. :note: If the directory containing the lock was removed, an exception will be raised during the blocking period, preventing hangs as the lock can never be obtained.""" __slots__ = ("_check_interval", "_max_block_time") def __init__( self, file_path: PathLike, check_interval_s: float = 0.3, max_block_time_s: int = maxsize, ) -> None: """Configure the instance :param check_interval_s: Period of time to sleep until the lock is checked the next time. By default, it waits a nearly unlimited time :param max_block_time_s: Maximum amount of seconds we may lock""" super(BlockingLockFile, self).__init__(file_path) self._check_interval = check_interval_s self._max_block_time = max_block_time_s def _obtain_lock(self) -> None: """This method blocks until it obtained the lock, or raises IOError if it ran out of time or if the parent directory was not available anymore. If this method returns, you are guaranteed to own the lock""" starttime = time.time() maxtime = starttime + float(self._max_block_time) while True: try: super(BlockingLockFile, self)._obtain_lock() except IOError as e: # synity check: if the directory leading to the lockfile is not # readable anymore, raise an exception curtime = time.time() if not osp.isdir(osp.dirname(self._lock_file_path())): msg = "Directory containing the lockfile %r was not readable anymore after waiting %g seconds" % ( self._lock_file_path(), curtime - starttime, ) raise IOError(msg) from e # END handle missing directory if curtime >= maxtime: msg = "Waited %g seconds for lock at %r" % ( maxtime - starttime, self._lock_file_path(), ) raise IOError(msg) from e # END abort if we wait too long time.sleep(self._check_interval) else: break # END endless loop class IterableList(List[T_IterableObj]): """ List of iterable objects allowing to query an object by id or by named index:: heads = repo.heads heads.master heads['master'] heads[0] Iterable parent objects = [Commit, SubModule, Reference, FetchInfo, PushInfo] Iterable via inheritance = [Head, TagReference, RemoteReference] ] It requires an id_attribute name to be set which will be queried from its contained items to have a means for comparison. A prefix can be specified which is to be used in case the id returned by the items always contains a prefix that does not matter to the user, so it can be left out.""" __slots__ = ("_id_attr", "_prefix") def __new__(cls, id_attr: str, prefix: str = "") -> "IterableList[IterableObj]": return super(IterableList, cls).__new__(cls) def __init__(self, id_attr: str, prefix: str = "") -> None: self._id_attr = id_attr self._prefix = prefix def __contains__(self, attr: object) -> bool: # first try identity match for performance try: rval = list.__contains__(self, attr) if rval: return rval except (AttributeError, TypeError): pass # END handle match # otherwise make a full name search try: getattr(self, cast(str, attr)) # use cast to silence mypy return True except (AttributeError, TypeError): return False # END handle membership def __getattr__(self, attr: str) -> T_IterableObj: attr = self._prefix + attr for item in self: if getattr(item, self._id_attr) == attr: return item # END for each item return list.__getattribute__(self, attr) def __getitem__(self, index: Union[SupportsIndex, int, slice, str]) -> T_IterableObj: # type: ignore assert isinstance(index, (int, str, slice)), "Index of IterableList should be an int or str" if isinstance(index, int): return list.__getitem__(self, index) elif isinstance(index, slice): raise ValueError("Index should be an int or str") else: try: return getattr(self, index) except AttributeError as e: raise IndexError("No item found with id %r" % (self._prefix + index)) from e # END handle getattr def __delitem__(self, index: Union[SupportsIndex, int, slice, str]) -> None: assert isinstance(index, (int, str)), "Index of IterableList should be an int or str" delindex = cast(int, index) if not isinstance(index, int): delindex = -1 name = self._prefix + index for i, item in enumerate(self): if getattr(item, self._id_attr) == name: delindex = i break # END search index # END for each item if delindex == -1: raise IndexError("Item with name %s not found" % name) # END handle error # END get index to delete list.__delitem__(self, delindex) class IterableClassWatcher(type): """Metaclass that watches""" def __init__(cls, name: str, bases: Tuple, clsdict: Dict) -> None: for base in bases: if type(base) == IterableClassWatcher: warnings.warn( f"GitPython Iterable subclassed by {name}. " "Iterable is deprecated due to naming clash since v3.1.18" " and will be removed in 3.1.20, " "Use IterableObj instead \n", DeprecationWarning, stacklevel=2, ) class Iterable(metaclass=IterableClassWatcher): """Defines an interface for iterable items which is to assure a uniform way to retrieve and iterate items within the git repository""" __slots__ = () _id_attribute_ = "attribute that most suitably identifies your instance" @classmethod def list_items(cls, repo: "Repo", args: Any, kwargs: Any) -> Any: """ Deprecated, use IterableObj instead. Find all items of this type - subclasses can specify args and kwargs differently. If no args are given, subclasses are obliged to return all items if no additional arguments arg given. :note: Favor the iter_items method as it will :return: list(Item,...) list of item instances""" out_list: Any = IterableList(cls._id_attribute_) out_list.extend(cls.iter_items(repo, args, *kwargs)) return out_list @classmethod def iter_items(cls, repo: "Repo", args: Any, *kwargs: Any) -> Any: # return typed to be compatible with subtypes e.g. Remote """For more information about the arguments, see list_items :return: iterator yielding Items""" raise NotImplementedError("To be implemented by Subclass") @runtime_checkable class IterableObj(Protocol): """Defines an interface for iterable items which is to assure a uniform way to retrieve and iterate items within the git repository Subclasses = [Submodule, Commit, Reference, PushInfo, FetchInfo, Remote]""" __slots__ = () _id_attribute_: str @classmethod def list_items(cls, repo: "Repo", args: Any, *kwargs: Any) -> IterableList[T_IterableObj]: """ Find all items of this type - subclasses can specify args and kwargs differently. If no args are given, subclasses are obliged to return all items if no additional arguments arg given. :note: Favor the iter_items method as it will :return: list(Item,...) list of item instances""" out_list: IterableList = IterableList(cls._id_attribute_) out_list.extend(cls.iter_items(repo, args, *kwargs)) return out_list @classmethod @abstractmethod def iter_items(cls, repo: "Repo", args: Any, **kwargs: Any) -> Iterator[T_IterableObj]: # Iterator[T_IterableObj]: # return typed to be compatible with subtypes e.g. Remote """For more information about the arguments, see list_items :return: iterator yielding Items""" raise NotImplementedError("To be implemented by Subclass") # } END classes class NullHandler(logging.Handler): def emit(self, record: object) -> None: pass	38.94 KB	2026-01-20 12:58

Terminal