finalizer=None, decode_streams=True):

"""Registers for notifications to lean 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, name, stream, is_decode, handler):

try:

for line in stream:

if handler:

if is_decode:

line = line.decode(defenc)

handler(line)

except Exception as ex:

log.error("Pumping %r of cmd(%s) failed due to: %r", name, cmdline, ex)

raise CommandError(['<%s-pump>' % name] + cmdline, 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.setDaemon(True)

t.start()

threads.append(t)

## FIXME: Why Join?? Will block if `stdin` needs feeding...

#

for t in threads:

t.join()

if finalizer:

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

return slots_to_dict(self, exclude=self._excluded_)

def __setstate__(self, d):

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

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

if sys.version_info < (3,):

# - a GitCommandNotFound error is spawned by ourselves

# - a OSError is spawned if the git executable provided

# cannot be executed for whatever reason

exceptions = (GitCommandNotFound, OSError)

else:

# - a GitCommandNotFound error is spawned by ourselves

# - a PermissionError is spawned if the git executable provided

# cannot be executed for whatever reason

exceptions = (GitCommandNotFound, PermissionError)

has_git = False

try:

cls().version()

has_git = True

except exceptions:

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

return is_cygwin_git(cls.GIT_PYTHON_GIT_EXECUTABLE)

@classmethod

def polish_url(cls, url, is_cygwin=None):

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 = 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, args):

self.proc = proc

self.args = args

def __del__(self):

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 ?

if proc.poll() is not None:

return

# can be that nothing really exists anymore ...

if os is None or getattr(os, 'kill', None) is None:

return

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

pass # ignore error when process already died

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

return getattr(self.proc, attr)

def wait(self, stderr=b''): # TODO: Bad choice to mimic `proc.wait()` but with different args.

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

if stderr is None:

stderr = b''

stderr = force_bytes(stderr)

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(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__ = ('_stream', '_nbr', '_size')

def __init__(self, size, stream):

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

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

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

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

def __iter__(self):

return self

def next(self):

line = self.readline()

if not line:

raise StopIteration

return line

def __del__(self):

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

"""Initialize this instance with:

super(Git, self).__init__()

self._working_dir = expand_path(working_dir)

self._git_options = ()

self._persistent_git_options = []

# Extra environment variables to pass to git commands

self._environment = {}

# cached command slots

self.cat_file_header = None

self.cat_file_all = None

def __getattr__(self, name):

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

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

if attr == '_version_info':

# We only use the first 4 numbers, as everything else could be strings in fact (on windows)

version_numbers = self._call_process('version').split(' ')[2]

self._version_info = 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):

""":return: Git directory we are working on"""

return self._working_dir

@property

def version_info(self):

"""

return self._version_info

def execute(self, command,

istream=None,

with_extended_output=False,

with_exceptions=True,

as_process=False,

output_stream=None,

stdout_as_string=True,

kill_after_timeout=None,

with_stdout=True,

universal_newlines=False,

shell=None,

env=None,

max_chunk_size=io.DEFAULT_BUFFER_SIZE,

**subprocess_kwargs

):

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

if self.GIT_PYTHON_TRACE and (self.GIT_PYTHON_TRACE != 'full' or as_process):

log.info(' '.join(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(command, '"kill_after_timeout" feature is not supported on Windows.')

else:

if sys.version_info[0] > 2:

cmd_not_found_exception = FileNotFoundError # NOQA # exists, flake8 unknown @UndefinedVariable

else:

cmd_not_found_exception = OSError

# end handle

stdout_sink = (PIPE

if with_stdout

else getattr(subprocess, 'DEVNULL', None) or open(os.devnull, 'wb'))

log.debug("Popen(%s, cwd=%s, universal_newlines=%s, shell=%s)",

command, cwd, universal_newlines, shell)

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(command, err)

if as_process:

return self.AutoInterrupt(proc, command)

def _kill_process(pid):

""" Callback method to kill a process. """

p = Popen(['ps', '--ppid', str(pid)], stdout=PIPE,

creationflags=PROC_CREATIONFLAGS)

child_pids = []

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 = b''

stderr_value = b''

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.isSet():

stderr_value = ('Timeout: the command "%s" did not complete in %d '

'secs.' % (" ".join(command), kill_after_timeout)).encode(defenc)

# strip trailing "\n"

if stdout_value.endswith(b"\n"):

stdout_value = stdout_value[:-1]

if stderr_value.endswith(b"\n"):

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(b"\n"):

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(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(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, value, split_single_char_options):

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 not in (False, None):

return ["--%s=%s" % (dashify(name), value)]

return []

def transform_kwargs(self, split_single_char_options=True, **kwargs):

"""Transforms Python style kwargs into git command line options."""

args = []

kwargs = OrderedDict(sorted(kwargs.items(), key=lambda x: x[0]))

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

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

# This is just required for unicode conversion, as subprocess can't handle it

# However, in any other case, passing strings (usually utf-8 encoded) is totally fine

if not PY3 and isinstance(arg_list, unicode):

return [arg_list.encode(defenc)]

return [str(arg_list)]

outlist = []

for arg in arg_list:

if isinstance(arg_list, (list, tuple)):

outlist.extend(cls.__unpack_args(arg))

elif not PY3 and isinstance(arg_list, unicode):

outlist.append(arg_list.encode(defenc))

# END recursion

else:

outlist.append(str(arg))

# END for each arg

return outlist

def __call__(self, **kwargs):

"""Specify command line options to the git executable

self._git_options = self.transform_kwargs(

split_single_char_options=True, **kwargs)

return self

def _call_process(self, method, *args, **kwargs):

"""Run the given git command with the specified arguments and return

# 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 = opt_args + ext_args

else:

try:

index = ext_args.index(insert_after_this_arg)

except ValueError:

raise ValueError("Couldn't find argument '%s' in args %s to insert cmd options after"

% (insert_after_this_arg, str(ext_args)))

# end handle error

args = ext_args[:index + 1] + opt_args + ext_args[index + 1:]

# end handle opts_kwargs