finalizer=None, decode_streams=True):

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

# Use 2 "pupm" 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 list(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

# The size in bytes read from stdout when copying git's output to another stream

max_chunk_size = io.DEFAULT_BUFFER_SIZE

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)

# Provide the full path to the git executable. Otherwise it assumes git is in the path

_git_exec_env_var = "GIT_PYTHON_GIT_EXECUTABLE"

GIT_PYTHON_GIT_EXECUTABLE = os.environ.get(_git_exec_env_var, git_exec_name)

# 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

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

# leave all additional logic to our readline method, we just check the size

out = list()

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

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

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

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

stream_copy(proc.stdout, output_stream, self.max_chunk_size)

stdout_value = output_stream

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 list(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 type(value) is not bool:

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 type(value) is not bool:

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

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

for k, v in list(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, str):

return [arg_list.encode(defenc)]

return [str(arg_list)]

outlist = list()

for arg in arg_list:

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

outlist.extend(cls.__unpack_args(arg))

elif not PY3 and isinstance(arg_list, str):

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 = dict((k, v) for k, v in list(kwargs.items()) if k in execute_kwargs)

opts_kwargs = dict((k, v) for k, v in list(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

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)

return self.execute(call, **exec_kwargs)

def _parse_object_header(self, header_line):

"""

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

# required for command to separate refs on stdin, as bytes

refstr = ref

if isinstance(ref, bytes):

# Assume 40 bytes hexsha - bin-to-ascii for some reason returns bytes, not text

refstr = ref.decode('ascii')

elif not isinstance(ref, string_types):

refstr = str(ref) # could be ref-object

if not refstr.endswith("\n"):

refstr += "\n"

return refstr.encode(defenc)

def _get_persistent_cmd(self, attr_name, cmd_name, *args, **kwargs):

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)

return cmd

def __get_object_header(self, cmd, ref):

cmd.stdin.write(self._prepare_ref(ref))

cmd.stdin.flush()

return self._parse_object_header(cmd.stdout.readline())

def get_object_header(self, ref):

""" Use this method to quickly examine the type and size of the object behind

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

""" As get_object_header, but returns object data as well

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

""" As get_object_header, but returns the data as a stream

cmd = self._get_persistent_cmd("cat_file_all", "cat_file", batch=True)

hexsha, typename, size = self.__get_object_header(cmd, ref)

return (hexsha, typename, size, self.CatFileContentStream(size, cmd.stdout))

def clear_cache(self):

"""Clear all kinds of internal caches to release resources.

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