diff options
Diffstat (limited to 'git/cmd.py')
| -rw-r--r-- | git/cmd.py | 225 |
1 files changed, 160 insertions, 65 deletions
@@ -42,12 +42,14 @@ from .util import ( # typing --------------------------------------------------------------------------- -from typing import Any, BinaryIO, Callable, Dict, List, Mapping, Sequence, TYPE_CHECKING, Union +from typing import (Any, AnyStr, BinaryIO, Callable, Dict, IO, List, Mapping, + Sequence, TYPE_CHECKING, Tuple, Union, cast, overload) -from git.types import PathLike, TBD +from git.types import PathLike, Literal, TBD if TYPE_CHECKING: - pass + from git.repo.base import Repo + from git.diff import DiffIndex # --------------------------------------------------------------------------------- @@ -70,10 +72,16 @@ __all__ = ('Git',) ## @{ def handle_process_output(process: subprocess.Popen, - stdout_handler: Union[None, Callable[[str], None]], - stderr_handler: Union[None, Callable[[str], None]], - finalizer: Union[None, Callable[[subprocess.Popen], TBD]] = None, - decode_streams: bool = True) -> Union[None, TBD]: # TBD is whatever finalizer returns + 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 the respective line handlers. This function returns once the finalizer returns @@ -327,8 +335,18 @@ class Git(LazyMixin): def is_cygwin(cls) -> bool: return is_cygwin_git(cls.GIT_PYTHON_GIT_EXECUTABLE) + @overload @classmethod def polish_url(cls, url: str, is_cygwin: Union[None, bool] = None) -> str: + ... + + @overload + @classmethod + def polish_url(cls, url: PathLike, is_cygwin: Union[None, bool] = None) -> PathLike: + ... + + @classmethod + def polish_url(cls, url: PathLike, is_cygwin: Union[None, bool] = None) -> PathLike: if is_cygwin is None: is_cygwin = cls.is_cygwin() @@ -443,7 +461,7 @@ class Git(LazyMixin): __slots__ = ('_stream', '_nbr', '_size') - def __init__(self, size: int, stream: BinaryIO) -> None: + def __init__(self, size: int, stream: IO[bytes]) -> None: self._stream = stream self._size = size self._nbr = 0 # num bytes read @@ -538,7 +556,7 @@ class Git(LazyMixin): self._stream.read(bytes_left + 1) # END handle incomplete read - def __init__(self, working_dir: Union[None, PathLike]=None): + def __init__(self, working_dir: Union[None, PathLike] = None): """Initialize this instance with: :param working_dir: @@ -548,17 +566,17 @@ class Git(LazyMixin): .git directory in case of bare repositories.""" super(Git, self).__init__() self._working_dir = expand_path(working_dir) - self._git_options = () - self._persistent_git_options = [] + self._git_options = () # type: Union[List[str], Tuple[str, ...]] + self._persistent_git_options = [] # type: List[str] # Extra environment variables to pass to git commands - self._environment = {} + self._environment = {} # type: Dict[str, str] # cached command slots self.cat_file_header = None self.cat_file_all = None - def __getattr__(self, name): + 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.""" @@ -566,7 +584,7 @@ class Git(LazyMixin): return LazyMixin.__getattr__(self, name) return lambda *args, **kwargs: self._call_process(name, *args, **kwargs) - def set_persistent_git_options(self, **kwargs): + def set_persistent_git_options(self, **kwargs: Any) -> None: """Specify command line options to the git executable for subsequent subcommand calls @@ -580,43 +598,96 @@ class Git(LazyMixin): self._persistent_git_options = self.transform_kwargs( split_single_char_options=True, **kwargs) - def _set_cache_(self, attr): + 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) - version_numbers = self._call_process('version').split(' ')[2] - self._version_info = tuple(int(n) for n in version_numbers.split('.')[:4] if n.isdigit()) + process_version = self._call_process('version') # should be as default *args and **kwargs used + version_numbers = process_version.split(' ')[2] + + self._version_info = tuple( + int(n) for n in version_numbers.split('.')[:4] if n.isdigit() + ) # type: Tuple[int, int, int, int] # type: ignore else: super(Git, self)._set_cache_(attr) # END handle version info @property - def working_dir(self): + def working_dir(self) -> Union[None, str]: """:return: Git directory we are working on""" return self._working_dir @property - def version_info(self): + 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 - 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 - ): + @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 the returned information (stdout) @@ -758,22 +829,31 @@ class Git(LazyMixin): creationflags=PROC_CREATIONFLAGS, **subprocess_kwargs ) + proc = cast(Popen[bytes], proc) + + proc.stdout = cast(BinaryIO, proc.stdout) except cmd_not_found_exception as err: raise GitCommandNotFound(redacted_command, err) from err + else: + assert isinstance(proc.stdout, BinaryIO) + assert isinstance(proc.stderr, BinaryIO) + # proc.stdout = cast(BinaryIO, proc.stdout) + # proc.stderr = cast(BinaryIO, proc.stderr) if as_process: return self.AutoInterrupt(proc, command) - def _kill_process(pid): + 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 = [] - 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)) + 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) @@ -797,8 +877,8 @@ class Git(LazyMixin): # Wait for the process to return status = 0 - stdout_value = b'' - stderr_value = b'' + stdout_value = b'' # type: Union[str, bytes] + stderr_value = b'' # type: Union[str, bytes] newline = "\n" if universal_newlines else b"\n" try: if output_stream is None: @@ -807,16 +887,17 @@ class Git(LazyMixin): stdout_value, stderr_value = proc.communicate() if kill_after_timeout: watchdog.cancel() - if kill_check.isSet(): + 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): + if stdout_value.endswith(newline): # type: ignore stdout_value = stdout_value[:-1] - if stderr_value.endswith(newline): + 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 @@ -824,7 +905,7 @@ class Git(LazyMixin): stdout_value = proc.stdout.read() stderr_value = proc.stderr.read() # strip trailing "\n" - if stderr_value.endswith(newline): + if stderr_value.endswith(newline): # type: ignore stderr_value = stderr_value[:-1] status = proc.wait() # END stdout handling @@ -908,7 +989,7 @@ class Git(LazyMixin): finally: self.update_environment(**old_env) - def transform_kwarg(self, name, value, split_single_char_options): + 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] @@ -924,7 +1005,7 @@ class Git(LazyMixin): return ["--%s=%s" % (dashify(name), value)] return [] - def transform_kwargs(self, split_single_char_options=True, **kwargs): + def transform_kwargs(self, split_single_char_options: bool = True, **kwargs: Any) -> List[str]: """Transforms Python style kwargs into git command line options.""" # Python 3.6 preserves the order of kwargs and thus has a stable # order. For older versions sort the kwargs by the key to get a stable @@ -943,7 +1024,7 @@ class Git(LazyMixin): return args @classmethod - def __unpack_args(cls, arg_list): + def __unpack_args(cls, arg_list: Sequence[str]) -> List[str]: if not isinstance(arg_list, (list, tuple)): return [str(arg_list)] @@ -957,7 +1038,7 @@ class Git(LazyMixin): # END for each arg return outlist - def __call__(self, **kwargs): + def __call__(self, **kwargs: Any) -> 'Git': """Specify command line options to the git executable for a subcommand call @@ -973,7 +1054,18 @@ class Git(LazyMixin): split_single_char_options=True, **kwargs) return self - def _call_process(self, method, *args, **kwargs): + @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, *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 @@ -1001,7 +1093,9 @@ class Git(LazyMixin): git rev-list max-count 10 --header master - :return: Same as ``execute``""" + :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} @@ -1010,11 +1104,12 @@ class Git(LazyMixin): 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 + args_list = opt_args + ext_args else: try: index = ext_args.index(insert_after_this_arg) @@ -1022,7 +1117,7 @@ class Git(LazyMixin): 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 = ext_args[:index + 1] + opt_args + ext_args[index + 1:] + args_list = ext_args[:index + 1] + opt_args + ext_args[index + 1:] # end handle opts_kwargs call = [self.GIT_PYTHON_GIT_EXECUTABLE] @@ -1036,11 +1131,11 @@ class Git(LazyMixin): self._git_options = () call.append(dashify(method)) - call.extend(args) + call.extend(args_list) return self.execute(call, **exec_kwargs) - def _parse_object_header(self, header_line): + def _parse_object_header(self, header_line: str) -> Tuple[str, str, int]: """ :param header_line: <hex_sha> type_string size_as_int @@ -1062,12 +1157,11 @@ class Git(LazyMixin): raise ValueError("Failed to parse header: %r" % header_line) return (tokens[0], tokens[1], int(tokens[2])) - def _prepare_ref(self, ref): + def _prepare_ref(self, ref: AnyStr) -> bytes: # 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') + refstr = ref.decode('ascii') # type: str elif not isinstance(ref, str): refstr = str(ref) # could be ref-object @@ -1075,7 +1169,8 @@ class Git(LazyMixin): refstr += "\n" return refstr.encode(defenc) - def _get_persistent_cmd(self, attr_name, cmd_name, *args, **kwargs): + def _get_persistent_cmd(self, attr_name: str, cmd_name: str, *args: Any, **kwargs: Any + ) -> Union['Git.AutoInterrupt', TBD]: cur_val = getattr(self, attr_name) if cur_val is not None: return cur_val @@ -1087,12 +1182,12 @@ class Git(LazyMixin): setattr(self, attr_name, cmd) return cmd - def __get_object_header(self, cmd, ref): + def __get_object_header(self, cmd, ref: AnyStr) -> Tuple[str, str, int]: cmd.stdin.write(self._prepare_ref(ref)) cmd.stdin.flush() return self._parse_object_header(cmd.stdout.readline()) - def get_object_header(self, ref): + def get_object_header(self, ref: AnyStr) -> Tuple[str, str, int]: """ Use this method to quickly examine the type and size of the object behind the given ref. @@ -1103,7 +1198,7 @@ class Git(LazyMixin): 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): + def get_object_data(self, ref: AnyStr) -> 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""" @@ -1112,7 +1207,7 @@ class Git(LazyMixin): del(stream) return (hexsha, typename, size, data) - def stream_object_data(self, ref): + def stream_object_data(self, ref: AnyStr) -> 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) @@ -1121,7 +1216,7 @@ class Git(LazyMixin): hexsha, typename, size = self.__get_object_header(cmd, ref) return (hexsha, typename, size, self.CatFileContentStream(size, cmd.stdout)) - def clear_cache(self): + def clear_cache(self) -> 'Git': """Clear all kinds of internal caches to release resources. Currently persistent commands will be interrupted. |