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[[subprocess.Popen], None]] = None,

decode_streams: bool = True) -> None:

"""Registers for notifications to learn that process output is ready to read, and dispatches lines to

# Use 2 "pump" threads and wait for both to finish.

def pump_stream(cmdline: 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("Pumping %r of cmd(%s) failed due to: %r", name, remove_password_if_present(cmdline), ex)

raise CommandError(['<%s-pump>' % name] + remove_password_if_present(cmdline), ex) from ex

finally:

stream.close()

cmdline = getattr(process, 'args', '') # PY3+ only

if not isinstance(cmdline, (tuple, list)):

cmdline = cmdline.split()

pumps = []

if process.stdout:

pumps.append(('stdout', process.stdout, stdout_handler))

if process.stderr:

pumps.append(('stderr', process.stderr, stderr_handler))

threads = []

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()

if finalizer:

return finalizer(process)

else:

for k, v in d.items():

setattr(self, k, v)

for k in excluded:

"""

__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')

def __getstate__(self) -> Dict[str, Any]:

return slots_to_dict(self, exclude=self._excluded_)

def __setstate__(self, d) -> 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

# 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("""\

""") % 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("""\

""") % (

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("""\

""") % (

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.

url = os.path.expandvars(url)

if url.startswith('~'):

url = os.path.expanduser(url)

url = url.replace("\\\\", "\\").replace("\\", "/")

return url

class AutoInterrupt(object):

"""Kill/Interrupt the stored process instance once this instance goes out of scope. It is

__slots__ = ("proc", "args")

def __init__(self, proc: Union[None, subprocess.Popen], args: Any) -> None:

self.proc = proc

self.args = args

def __del__(self) -> None:

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:

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()

proc.wait() # ensure process goes away

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 __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, bytes] = b'') -> int:

"""Wait for the process and return its status code.

if stderr is None:

stderr = b''

stderr = force_bytes(data=stderr, encoding='utf-8')

if self.proc is not None:

status = self.proc.wait()

def read_all_from_possibly_closed_stream(stream):

try:

return stderr + force_bytes(stream.read())

except ValueError:

return stderr or b''

if status != 0:

errstr = read_all_from_possibly_closed_stream(self.proc.stderr)

log.debug('AutoInterrupt wait stderr: %r' % (errstr,))

raise GitCommandError(remove_password_if_present(self.args), status, errstr)

# END status handling

return status

# END auto interrupt

class CatFileContentStream(object):

"""Object representing a sized read-only stream returning the contents of

__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:

return next(self)

def next(self) -> bytes:

line = self.readline()

if not line:

raise StopIteration

return line

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:

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 = None

self.cat_file_all = None

def __getattr__(self, name: str) -> Any:

"""A convenience method as it allows to call the command as if it was

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

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 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, int] = 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,

**subprocess_kwargs: Any

) -> Union[str, bytes, Tuple[int, Union[str, bytes], str], AutoInterrupt]:

"""Handles executing the command on the shell and consumes and returns

# 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.

cwd = self._working_dir or os.getcwd()

# 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:

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,

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:

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:

watchdog.start()

stdout_value, stderr_value = proc.communicate()

if kill_after_timeout:

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): # 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):

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):

return self._environment

def update_environment(self, **kwargs):

"""

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):

"""

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."""