Add types to cmd.py Git

Author: Yobmod

git/cmd.py

@@ -42,12 +42,12 @@
 
 # typing ---------------------------------------------------------------------------
 
-from typing import TYPE_CHECKING
+from typing import Any, BinaryIO, Callable, Dict, Mapping, Sequence, TYPE_CHECKING, Union
 
-from git.types import TBD
+from git.types import PathLike, TBD
 
 if TYPE_CHECKING:
-   pass
+    pass
 
 
 # ---------------------------------------------------------------------------------
@@ -69,8 +69,11 @@
 # Documentation
 ## @{
 
-def handle_process_output(process, stdout_handler, stderr_handler,
-                          finalizer=None, decode_streams=True):
+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
     """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
@@ -87,13 +90,14 @@ def handle_process_output(process, stdout_handler, stderr_handler,
         or if decoding must happen later (i.e. for Diffs).
     """
     # Use 2 "pump" threads and wait for both to finish.
-    def pump_stream(cmdline, name, stream, is_decode, handler):
+    def pump_stream(cmdline: str, name: str, stream: BinaryIO, is_decode: bool,
+                    handler: Union[None, Callable[[str], None]]) -> None:
         try:
             for line in stream:
                 if handler:
                     if is_decode:
-                        line = line.decode(defenc)
-                    handler(line)
+                        line_str = line.decode(defenc)
+                    handler(line_str)
         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
@@ -126,17 +130,20 @@ def pump_stream(cmdline, name, stream, is_decode, handler):
 
     if finalizer:
         return finalizer(process)
+    else:
+        return None
 
 
-def dashify(string):
+def dashify(string: str) -> str:
     return string.replace('_', '-')
 
 
-def slots_to_dict(self, exclude=()):
+def slots_to_dict(self, exclude: Sequence[str] = ()) -> Dict[str, Any]:
+    # annotate self.__slots__ as Tuple[str, ...] once 3.5 dropped
     return {s: getattr(self, s) for s in self.__slots__ if s not in exclude}
 
 
-def dict_to_slots_and__excluded_are_none(self, d, excluded=()):
+def dict_to_slots_and__excluded_are_none(self, d: Mapping[str, Any], excluded: Sequence[str] = ()) -> None:
     for k, v in d.items():
         setattr(self, k, v)
     for k in excluded:
@@ -175,10 +182,10 @@ class Git(LazyMixin):
 
     _excluded_ = ('cat_file_all', 'cat_file_header', '_version_info')
 
-    def __getstate__(self):
+    def __getstate__(self) -> Dict[str, Any]:
         return slots_to_dict(self, exclude=self._excluded_)
 
-    def __setstate__(self, d):
+    def __setstate__(self, d) -> None:
         dict_to_slots_and__excluded_are_none(self, d, excluded=self._excluded_)
 
     # CONFIGURATION
@@ -202,7 +209,7 @@ def __setstate__(self, d):
     # the top level __init__
 
     @classmethod
-    def refresh(cls, path=None):
+    def refresh(cls, path: Union[None, PathLike] = None) -> bool:
         """This gets called by the refresh function (see the top level
         __init__).
         """
@@ -317,11 +324,11 @@ def refresh(cls, path=None):
         return has_git
 
     @classmethod
-    def is_cygwin(cls):
+    def is_cygwin(cls) -> bool:
         return is_cygwin_git(cls.GIT_PYTHON_GIT_EXECUTABLE)
 
     @classmethod
-    def polish_url(cls, url, is_cygwin=None):
+    def polish_url(cls, url: str, is_cygwin: Union[None, bool] = None) -> str:
         if is_cygwin is None:
             is_cygwin = cls.is_cygwin()
 
@@ -338,7 +345,6 @@ def polish_url(cls, url, is_cygwin=None):
             if url.startswith('~'):
                 url = os.path.expanduser(url)
             url = url.replace("\\\\", "\\").replace("\\", "/")
-
         return url
 
     class AutoInterrupt(object):

git/util.py

@@ -22,11 +22,13 @@
 # typing ---------------------------------------------------------
 
 from typing import (Any, AnyStr, BinaryIO, Callable, Dict, Generator, IO, Iterator, List,
-                    Optional, Pattern, Sequence, Tuple, Union, cast, TYPE_CHECKING)
+                    Optional, Pattern, Sequence, Tuple, Union, cast, TYPE_CHECKING, overload)
+
+
 if TYPE_CHECKING:
     from git.remote import Remote
     from git.repo.base import Repo
-from .types import PathLike, TBD
+from .types import PathLike, TBD, Literal
 
 # ---------------------------------------------------------------------
 
@@ -279,9 +281,20 @@ def _cygexpath(drive: Optional[str], path: PathLike) -> str:
 )  # type: Tuple[Tuple[Pattern[str], Callable, bool], ...]
 
 
+@overload
+def cygpath(path: str) -> str:
+    ...
+
+
+@overload
+def cygpath(path: PathLike) -> PathLike:
+    ...
+
+
 def cygpath(path: PathLike) -> PathLike:
     """Use :meth:`git.cmd.Git.polish_url()` instead, that works on any environment."""
-    path = str(path)  # ensure is str and not AnyPath
+    path = str(path)  # ensure is str and not AnyPath.
+    #Fix to use Paths when 3.5 dropped. or to be just str if only for urls?
     if not path.startswith(('/cygdrive', '//')):
         for regex, parser, recurse in _cygpath_parsers:
             match = regex.match(path)
@@ -314,10 +327,23 @@ def decygpath(path: PathLike) -> str:
 _is_cygwin_cache = {}  # type: Dict[str, Optional[bool]]
 
 
+@overload
+def is_cygwin_git(git_executable: None) -> Literal[False]:
+    ...
+
+
+@overload
 def is_cygwin_git(git_executable: PathLike) -> bool:
+    ...
+
+
+def is_cygwin_git(git_executable: Union[None, PathLike]) -> bool:
     if not is_win:
         return False
 
+    if git_executable is None:
+        return False  # or raise error?
+
     #from subprocess import check_output
     git_executable = str(git_executable)
     is_cygwin = _is_cygwin_cache.get(git_executable)  # type: Optional[bool]