Change remaining type comments to py3.6+ types

Author: Yobmod

.flake8

@@ -19,7 +19,7 @@ enable-extensions = TC, TC2  # only needed for extensions not enabled by default
 ignore = E265,E266,E731,E704,
          W293, W504,
          ANN0 ANN1 ANN2,
-         TC0, TC1, TC2
+         # TC0, TC1, TC2
          # B,
          A,
          D,

git/cmd.py

@@ -565,11 +565,11 @@ def __init__(self, working_dir: Union[None, PathLike] = None):
            .git directory in case of bare repositories."""
         super(Git, self).__init__()
         self._working_dir = expand_path(working_dir)
-        self._git_options = ()  # type: Union[List[str], Tuple[str, ...]]
-        self._persistent_git_options = []  # type: List[str]
+        self._git_options: Union[List[str], Tuple[str, ...]] = ()
+        self._persistent_git_options: List[str] = []
 
         # Extra environment variables to pass to git commands
-        self._environment = {}  # type: Dict[str, str]
+        self._environment: Dict[str, str] = {}
 
         # cached command slots
         self.cat_file_header = None
@@ -603,9 +603,9 @@ def _set_cache_(self, attr: str) -> None:
             process_version = self._call_process('version')  # should be as default *args and **kwargs used
             version_numbers = process_version.split(' ')[2]
 
-            self._version_info = tuple(
+            self._version_info: Tuple[int, int, int, int] = tuple(
                 int(n) for n in version_numbers.split('.')[:4] if n.isdigit()
-            )  # type: Tuple[int, int, int, int]  # type: ignore
+            )  # type: ignore  # use typeguard here
         else:
             super(Git, self)._set_cache_(attr)
         # END handle version info
@@ -868,8 +868,8 @@ def _kill_process(pid: int) -> None:
 
         # Wait for the process to return
         status = 0
-        stdout_value = b''  # type: Union[str, bytes]
-        stderr_value = b''  # type: Union[str, bytes]
+        stdout_value: Union[str, bytes] = b''
+        stderr_value: Union[str, bytes] = b''
         newline = "\n" if universal_newlines else b"\n"
         try:
             if output_stream is None:
@@ -1145,7 +1145,7 @@ def _prepare_ref(self, ref: AnyStr) -> bytes:
         # required for command to separate refs on stdin, as bytes
         if isinstance(ref, bytes):
             # Assume 40 bytes hexsha - bin-to-ascii for some reason returns bytes, not text
-            refstr = ref.decode('ascii')  # type: str
+            refstr: str = ref.decode('ascii')
         elif not isinstance(ref, str):
             refstr = str(ref)               # could be ref-object
         else:

git/compat.py

@@ -34,7 +34,7 @@
 # ---------------------------------------------------------------------------
 
 
-is_win = (os.name == 'nt')  # type: bool
+is_win: bool = (os.name == 'nt')
 is_posix = (os.name == 'posix')
 is_darwin = (os.name == 'darwin')
 defenc = sys.getfilesystemencoding()

git/config.py

@@ -208,7 +208,7 @@ def get(self, key: str, default: Union[Any, None] = None) -> Union[Any, None]:
     def getall(self, key: str) -> Any:
         return super(_OMD, self).__getitem__(key)
 
-    def items(self) -> List[Tuple[str, Any]]:  # type: ignore  ## mypy doesn't like overwriting supertype signitures
+    def items(self) -> List[Tuple[str, Any]]:  # type: ignore[override]
         """List of (key, last value for key)."""
         return [(k, self[k]) for k in self]
 
@@ -322,7 +322,7 @@ def __init__(self, file_or_files: Union[None, PathLike, 'BytesIO', Sequence[Unio
         self._is_initialized = False
         self._merge_includes = merge_includes
         self._repo = repo
-        self._lock = None  # type: Union['LockFile', None]
+        self._lock: Union['LockFile', None] = None
         self._acquire_lock()
 
     def _acquire_lock(self) -> None:
@@ -545,13 +545,15 @@ def read(self) -> None:
             return None
         self._is_initialized = True
 
-        files_to_read = [""]  # type: List[Union[PathLike, IO]]  ## just for types until 3.5 dropped
-        if isinstance(self._file_or_files, (str)):  # replace with PathLike once 3.5 dropped
-            files_to_read = [self._file_or_files]                               # for str, as str is a type of Sequence
+        files_to_read: List[Union[PathLike, IO]] = [""]
+        if isinstance(self._file_or_files, (str, os.PathLike)):
+            # for str or Path, as str is a type of Sequence
+            files_to_read = [self._file_or_files]
         elif not isinstance(self._file_or_files, (tuple, list, Sequence)):
-            files_to_read = [self._file_or_files]                               # for IO or Path
-        else:
-            files_to_read = list(self._file_or_files)                           # for lists or tuples
+            # could merge with above isinstance once runtime type known
+            files_to_read = [self._file_or_files]
+        else:    # for lists or tuples
+            files_to_read = list(self._file_or_files)
         # end assure we have a copy of the paths to handle
 
         seen = set(files_to_read)

git/diff.py

@@ -351,13 +351,13 @@ def __hash__(self) -> int:
         return hash(tuple(getattr(self, n) for n in self.__slots__))
 
     def __str__(self) -> str:
-        h = "%s"  # type: str
+        h: str = "%s"
         if self.a_blob:
             h %= self.a_blob.path
         elif self.b_blob:
             h %= self.b_blob.path
 
-        msg = ''  # type: str
+        msg: str = ''
         line = None          # temp line
         line_length = 0      # line length
         for b, n in zip((self.a_blob, self.b_blob), ('lhs', 'rhs')):
@@ -449,7 +449,7 @@ def _index_from_patch_format(cls, repo: 'Repo', proc: TBD) -> DiffIndex:
         :return: git.DiffIndex """
 
         ## FIXME: Here SLURPING raw, need to re-phrase header-regexes linewise.
-        text_list = []  # type: List[bytes]
+        text_list: List[bytes] = []
         handle_process_output(proc, text_list.append, None, finalize_process, decode_streams=False)
 
         # for now, we have to bake the stream

git/index/base.py

@@ -123,7 +123,7 @@ def __init__(self, repo: 'Repo', file_path: Union[PathLike, None] = None) -> Non
         self.repo = repo
         self.version = self._VERSION
         self._extension_data = b''
-        self._file_path = file_path or self._index_path()  # type: PathLike
+        self._file_path: PathLike = file_path or self._index_path()
 
     def _set_cache_(self, attr: str) -> None:
         if attr == "entries":
@@ -136,7 +136,7 @@ def _set_cache_(self, attr: str) -> None:
                 ok = True
             except OSError:
                 # in new repositories, there may be no index, which means we are empty
-                self.entries = {}  # type: Dict[Tuple[PathLike, StageType], IndexEntry]
+                self.entries: Dict[Tuple[PathLike, StageType], IndexEntry] = {}
                 return None
             finally:
                 if not ok:
@@ -266,7 +266,7 @@ def merge_tree(self, rhs: Treeish, base: Union[None, Treeish] = None) -> 'IndexF
         # -i : ignore working tree status
         # --aggressive : handle more merge cases
         # -m : do an actual merge
-        args = ["--aggressive", "-i", "-m"]  # type: List[Union[Treeish, str]]
+        args: List[Union[Treeish, str]] = ["--aggressive", "-i", "-m"]
         if base is not None:
             args.append(base)
         args.append(rhs)
@@ -288,14 +288,14 @@ def new(cls, repo: 'Repo', *tree_sha: Union[str, Tree]) -> 'IndexFile':
             New IndexFile instance. Its path will be undefined.
             If you intend to write such a merged Index, supply an alternate file_path
             to its 'write' method."""
-        tree_sha_bytes = [to_bin_sha(str(t)) for t in tree_sha]  # List[bytes]
+        tree_sha_bytes: List[bytes] = [to_bin_sha(str(t)) for t in tree_sha]
         base_entries = aggressive_tree_merge(repo.odb, tree_sha_bytes)
 
         inst = cls(repo)
         # convert to entries dict
-        entries = dict(zip(
+        entries: Dict[Tuple[PathLike, int], IndexEntry] = dict(zip(
             ((e.path, e.stage) for e in base_entries),
-            (IndexEntry.from_base(e) for e in base_entries)))  # type: Dict[Tuple[PathLike, int], IndexEntry]
+            (IndexEntry.from_base(e) for e in base_entries)))
 
         inst.entries = entries
         return inst
@@ -338,7 +338,7 @@ def from_tree(cls, repo: 'Repo', *treeish: Treeish, **kwargs: Any) -> 'IndexFile
         if len(treeish) == 0 or len(treeish) > 3:
             raise ValueError("Please specify between 1 and 3 treeish, got %i" % len(treeish))
 
-        arg_list = []  # type: List[Union[Treeish, str]]
+        arg_list: List[Union[Treeish, str]] = []
         # ignore that working tree and index possibly are out of date
         if len(treeish) > 1:
             # drop unmerged entries when reading our index and merging
@@ -445,7 +445,7 @@ def _write_path_to_stdin(self, proc: 'Popen', filepath: PathLike, item: TBD, fma
             we will close stdin to break the pipe."""
 
         fprogress(filepath, False, item)
-        rval = None  # type: Union[None, str]
+        rval: Union[None, str] = None
 
         if proc.stdin is not None:
             try:
@@ -492,7 +492,7 @@ def unmerged_blobs(self) -> Dict[PathLike, List[Tuple[StageType, Blob]]]:
             are at stage 3 will not have a stage 3 entry.
         """
         is_unmerged_blob = lambda t: t[0] != 0
-        path_map = {}  # type: Dict[PathLike, List[Tuple[TBD, Blob]]]
+        path_map: Dict[PathLike, List[Tuple[TBD, Blob]]] = {}
         for stage, blob in self.iter_blobs(is_unmerged_blob):
             path_map.setdefault(blob.path, []).append((stage, blob))
         # END for each unmerged blob
@@ -624,8 +624,8 @@ def _store_path(self, filepath: PathLike, fprogress: Callable) -> BaseIndexEntry
         st = os.lstat(filepath)     # handles non-symlinks as well
         if S_ISLNK(st.st_mode):
             # in PY3, readlink is string, but we need bytes. In PY2, it's just OS encoded bytes, we assume UTF-8
-            open_stream = lambda: BytesIO(force_bytes(os.readlink(filepath),
-                                                      encoding=defenc))  # type: Callable[[], BinaryIO]
+            open_stream: Callable[[], BinaryIO] = lambda: BytesIO(force_bytes(os.readlink(filepath),
+                                                                              encoding=defenc))
         else:
             open_stream = lambda: open(filepath, 'rb')
         with open_stream() as stream:
@@ -1160,7 +1160,7 @@ def handle_stderr(proc: 'Popen[bytes]', iter_checked_out_files: Iterable[PathLik
             proc = self.repo.git.checkout_index(args, **kwargs)
             # FIXME: Reading from GIL!
             make_exc = lambda: GitCommandError(("git-checkout-index",) + tuple(args), 128, proc.stderr.read())
-            checked_out_files = []  # type: List[PathLike]
+            checked_out_files: List[PathLike] = []
 
             for path in paths:
                 co_path = to_native_path_linux(self._to_relative_path(path))

git/index/fun.py

@@ -99,8 +99,8 @@ def run_commit_hook(name: str, index: 'IndexFile', *args: str) -> None:
     except Exception as ex:
         raise HookExecutionError(hp, ex) from ex
     else:
-        stdout_list = []  # type:  List[str]
-        stderr_list = []  # type:  List[str]
+        stdout_list: List[str] = []
+        stderr_list: List[str] = []
         handle_process_output(cmd, stdout_list.append, stderr_list.append, finalize_process)
         stdout = ''.join(stdout_list)
         stderr = ''.join(stderr_list)
@@ -151,8 +151,8 @@ def write_cache(entries: Sequence[Union[BaseIndexEntry, 'IndexEntry']], stream:
         beginoffset = tell()
         write(entry[4])         # ctime
         write(entry[5])         # mtime
-        path_str = entry[3]  # type: str
-        path = force_bytes(path_str, encoding=defenc)
+        path_str: str = entry[3]
+        path: bytes = force_bytes(path_str, encoding=defenc)
         plen = len(path) & CE_NAMEMASK      # path length
         assert plen == len(path), "Path %s too long to fit into index" % entry[3]
         flags = plen | (entry[2] & CE_NAMEMASK_INV)     # clear possible previous values
@@ -210,7 +210,7 @@ def read_cache(stream: IO[bytes]) -> Tuple[int, Dict[Tuple[PathLike, int], 'Inde
     * content_sha is a 20 byte sha on all cache file contents"""
     version, num_entries = read_header(stream)
     count = 0
-    entries = {}  # type: Dict[Tuple[PathLike, int], 'IndexEntry']
+    entries: Dict[Tuple[PathLike, int], 'IndexEntry'] = {}
 
     read = stream.read
     tell = stream.tell

git/objects/submodule/base.py

@@ -475,7 +475,8 @@ def add(cls, repo: 'Repo', name: str, path: PathLike, url: Union[str, None] = No
                 sm._branch_path = br.path
 
         # we deliberately assume that our head matches our index !
-        sm.binsha = mrepo.head.commit.binsha  # type: ignore
+        if mrepo:
+            sm.binsha = mrepo.head.commit.binsha
         index.add([sm], write=True)
 
         return sm

git/objects/tag.py

@@ -51,7 +51,7 @@ def __init__(self, repo: 'Repo', binsha: bytes,
             authored_date is in, in a format similar to time.altzone"""
         super(TagObject, self).__init__(repo, binsha)
         if object is not None:
-            self.object = object    # type: Union['Commit', 'Blob', 'Tree', 'TagObject']
+            self.object: Union['Commit', 'Blob', 'Tree', 'TagObject'] = object
         if tag is not None:
             self.tag = tag
         if tagger is not None:
@@ -67,7 +67,7 @@ def _set_cache_(self, attr: str) -> None:
         """Cache all our attributes at once"""
         if attr in TagObject.__slots__:
             ostream = self.repo.odb.stream(self.binsha)
-            lines = ostream.read().decode(defenc, 'replace').splitlines()  # type: List[str]
+            lines: List[str] = ostream.read().decode(defenc, 'replace').splitlines()
 
             _obj, hexsha = lines[0].split(" ")
             _type_token, type_name = lines[1].split(" ")

git/objects/tree.py

@@ -298,7 +298,7 @@ def cache(self) -> TreeModifier:
             See the ``TreeModifier`` for more information on how to alter the cache"""
         return TreeModifier(self._cache)
 
-    def traverse(self,      # type: ignore  # overrides super()
+    def traverse(self,      # type: ignore[override]
                  predicate: Callable[[Union[IndexObjUnion, TraversedTreeTup], int], bool] = lambda i, d: True,
                  prune: Callable[[Union[IndexObjUnion, TraversedTreeTup], int], bool] = lambda i, d: False,
                  depth: int = -1,