diff options
| -rw-r--r-- | git/cmd.py | 225 | ||||
| -rw-r--r-- | git/compat.py | 4 | ||||
| -rw-r--r-- | git/diff.py | 10 | ||||
| -rw-r--r-- | git/exc.py | 13 | ||||
| -rw-r--r-- | git/util.py | 25 |
5 files changed, 194 insertions, 83 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. diff --git a/git/compat.py b/git/compat.py index 4ecd19a9..cbb39fa6 100644 --- a/git/compat.py +++ b/git/compat.py @@ -44,9 +44,9 @@ defenc = sys.getfilesystemencoding() def safe_decode(s: None) -> None: ... @overload -def safe_decode(s: Union[IO[str], AnyStr]) -> str: ... +def safe_decode(s: AnyStr) -> str: ... -def safe_decode(s: Union[IO[str], AnyStr, None]) -> Optional[str]: +def safe_decode(s: Union[AnyStr, None]) -> Optional[str]: """Safely decodes a binary string to unicode""" if isinstance(s, str): return s diff --git a/git/diff.py b/git/diff.py index 5a7b189f..ca673b0c 100644 --- a/git/diff.py +++ b/git/diff.py @@ -22,6 +22,8 @@ if TYPE_CHECKING: from .objects.tree import Tree from git.repo.base import Repo + from subprocess import Popen + Lit_change_type = Literal['A', 'D', 'M', 'R', 'T'] # ------------------------------------------------------------------------ @@ -490,7 +492,7 @@ class Diff(object): return index @staticmethod - def _handle_diff_line(lines_bytes: bytes, repo: 'Repo', index: TBD) -> None: + def _handle_diff_line(lines_bytes: bytes, repo: 'Repo', index: DiffIndex) -> None: lines = lines_bytes.decode(defenc) for line in lines.split(':')[1:]: @@ -542,14 +544,14 @@ class Diff(object): index.append(diff) @classmethod - def _index_from_raw_format(cls, repo: 'Repo', proc: TBD) -> DiffIndex: + 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() - handle_process_output(proc, lambda bytes: cls._handle_diff_line( - bytes, repo, index), None, finalize_process, decode_streams=False) + handle_process_output(proc, lambda byt: cls._handle_diff_line(byt, repo, index), + None, finalize_process, decode_streams=False) return index @@ -11,7 +11,7 @@ from git.compat import safe_decode # typing ---------------------------------------------------- -from typing import IO, List, Optional, Tuple, Union, TYPE_CHECKING +from typing import List, Optional, Tuple, Union, TYPE_CHECKING from git.types import PathLike if TYPE_CHECKING: @@ -49,8 +49,9 @@ class CommandError(GitError): _msg = "Cmd('%s') failed%s" def __init__(self, command: Union[List[str], Tuple[str, ...], str], - status: Union[str, None, Exception] = None, - stderr: Optional[IO[str]] = None, stdout: Optional[IO[str]] = None) -> None: + 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 = command @@ -92,8 +93,8 @@ class GitCommandError(CommandError): def __init__(self, command: Union[List[str], Tuple[str, ...], str], status: Union[str, int, None, Exception] = None, - stderr: Optional[IO[str]] = None, - stdout: Optional[IO[str]] = None, + stderr: Union[bytes, str, None] = None, + stdout: Union[bytes, str, None] = None, ) -> None: super(GitCommandError, self).__init__(command, status, stderr, stdout) @@ -139,7 +140,7 @@ class HookExecutionError(CommandError): via standard output""" def __init__(self, command: Union[List[str], Tuple[str, ...], str], status: Optional[str], - stderr: Optional[IO[str]] = None, stdout: Optional[IO[str]] = None) -> None: + stderr: Optional[str] = None, stdout: Optional[str] = None) -> None: super(HookExecutionError, self).__init__(command, status, stderr, stdout) self._msg = "Hook('%s') failed%s" diff --git a/git/util.py b/git/util.py index 76ac92f1..d1ea4c10 100644 --- a/git/util.py +++ b/git/util.py @@ -374,18 +374,31 @@ def get_user_id() -> str: return "%s@%s" % (getpass.getuser(), platform.node()) -def finalize_process(proc: TBD, **kwargs: Any) -> None: +def finalize_process(proc: subprocess.Popen, **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) -def expand_path(p: PathLike, expand_vars: bool = True) -> Optional[PathLike]: +@overload +def expand_path(p: None, expand_vars: bool = ...) -> None: + ... + + +@overload +def expand_path(p: PathLike, expand_vars: bool = ...) -> str: + ... + + +def expand_path(p: Union[None, PathLike], expand_vars: bool = True) -> Optional[str]: try: - p = osp.expanduser(p) - if expand_vars: - p = osp.expandvars(p) - return osp.normpath(osp.abspath(p)) + if p is not None: + p_out = osp.expanduser(p) + if expand_vars: + p_out = osp.expandvars(p_out) + return osp.normpath(osp.abspath(p_out)) + else: + return None except Exception: return None |