review

- also fixup typing

Assisted-by: GPT 6.0
Co-authored-by: GPT 6.0 <codex@openai.com>

Author: Byron

.basedpyright/baseline.json

@@ -100,7 +100,7 @@
 
 
 def handle_process_output(
-    process: "Git.AutoInterrupt" | Popen,
+    process: Union["Git.AutoInterrupt", Popen],
     stdout_handler: Union[
         None,
         Callable[[AnyStr], None],
@@ -395,9 +395,7 @@ def wait(self, stderr: Union[None, str, bytes] = b"") -> int:
         :raise git.exc.GitCommandError:
             If the return status is not 0.
         """
-        if stderr is None:
-            stderr_b = b""
-        stderr_b = force_bytes(data=stderr, encoding="utf-8")
+        stderr_b = force_bytes(data=stderr, encoding="utf-8") or b""
         status: Union[int, None]
         if self.proc is not None:
             status = self.proc.wait()
@@ -1180,52 +1178,112 @@ def version_info(self) -> Tuple[int, ...]:
     def execute(
         self,
         command: Union[str, Sequence[Any]],
+        istream: Union[None, int, BinaryIO] = None,
         *,
         as_process: Literal[True],
+        **subprocess_kwargs: Any,
     ) -> "AutoInterrupt": ...
 
     @overload
     def execute(
         self,
         command: Union[str, Sequence[Any]],
+        istream: Union[None, int, BinaryIO] = None,
         *,
         as_process: Literal[False] = False,
-        stdout_as_string: Literal[True],
-    ) -> Union[str, Tuple[int, str, str]]: ...
+        with_extended_output: Literal[False] = False,
+        stdout_as_string: Literal[True] = True,
+        with_stdout: Literal[True] = True,
+        **subprocess_kwargs: Any,
+    ) -> str: ...
 
     @overload
     def execute(
         self,
         command: Union[str, Sequence[Any]],
+        istream: Union[None, int, BinaryIO] = None,
         *,
         as_process: Literal[False] = False,
-        stdout_as_string: Literal[False] = False,
-    ) -> Union[bytes, Tuple[int, bytes, str]]: ...
+        with_extended_output: Literal[False] = False,
+        stdout_as_string: Literal[False],
+        universal_newlines: Literal[False] = False,
+        with_stdout: Literal[True] = True,
+        **subprocess_kwargs: Any,
+    ) -> bytes: ...
 
     @overload
     def execute(
         self,
         command: Union[str, Sequence[Any]],
+        istream: Union[None, int, BinaryIO] = None,
         *,
-        with_extended_output: Literal[False],
-        as_process: Literal[False],
-        stdout_as_string: Literal[True],
-    ) -> str: ...
+        as_process: Literal[False] = False,
+        with_extended_output: Literal[True],
+        stdout_as_string: Literal[True] = True,
+        with_stdout: Literal[True] = True,
+        **subprocess_kwargs: Any,
+    ) -> Tuple[int, str, str]: ...
 
     @overload
     def execute(
         self,
         command: Union[str, Sequence[Any]],
+        istream: Union[None, int, BinaryIO] = None,
         *,
-        with_extended_output: Literal[False],
-        as_process: Literal[False],
+        as_process: Literal[False] = False,
+        with_extended_output: Literal[True],
         stdout_as_string: Literal[False],
-    ) -> bytes: ...
+        universal_newlines: Literal[False] = False,
+        with_stdout: Literal[True] = True,
+        **subprocess_kwargs: Any,
+    ) -> Tuple[int, bytes, str]: ...
+
+    @overload
+    def execute(
+        self,
+        command: Union[str, Sequence[Any]],
+        istream: Union[None, int, BinaryIO] = None,
+        *,
+        as_process: Literal[False] = False,
+        with_extended_output: Literal[True],
+        **subprocess_kwargs: Any,
+    ) -> Tuple[int, Union[str, bytes, None], str]: ...
+
+    @overload
+    def execute(
+        self,
+        command: Union[str, Sequence[Any]],
+        istream: Union[None, int, BinaryIO] = None,
+        *,
+        as_process: Literal[False] = False,
+        with_extended_output: Literal[False] = False,
+        **subprocess_kwargs: Any,
+    ) -> Union[str, bytes, None]: ...
+
+    @overload
+    def execute(
+        self,
+        command: Union[str, Sequence[Any]],
+        istream: Union[None, int, 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, float] = 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,
+        strip_newline_in_stdout: bool = True,
+        **subprocess_kwargs: Any,
+    ) -> Union[None, str, bytes, Tuple[int, Union[str, bytes, None], str], AutoInterrupt]: ...
 
     def execute(
         self,
         command: Union[str, Sequence[Any]],
-        istream: Union[None, BinaryIO] = None,
+        istream: Union[None, int, BinaryIO] = None,
         with_extended_output: bool = False,
         with_exceptions: bool = True,
         as_process: bool = False,
@@ -1239,7 +1297,7 @@ def execute(
         max_chunk_size: int = io.DEFAULT_BUFFER_SIZE,
         strip_newline_in_stdout: bool = True,
         **subprocess_kwargs: Any,
-    ) -> Union[str, bytes, Tuple[int, Union[str, bytes], str], AutoInterrupt]:
+    ) -> Union[None, str, bytes, Tuple[int, Union[str, bytes, None], str], AutoInterrupt]:
         R"""Handle executing the command, and consume and return the returned
         information (stdout).
 
@@ -1503,7 +1561,7 @@ def make_timeout_error() -> Union[str, bytes]:
                 err = f'Timeout: the command "{" ".join(redacted_command)}" did not complete in {timeout:g} secs.'
                 return err if universal_newlines else err.encode(defenc)
 
-            def communicate() -> Tuple[AnyStr, AnyStr]:
+            def communicate() -> Tuple[Union[str, bytes, None], Union[str, bytes, None]]:
                 assert watchdog is not None
                 assert kill_check is not None
                 watchdog.start()
@@ -1523,8 +1581,8 @@ def communicate() -> Tuple[AnyStr, AnyStr]:
 
         # Wait for the process to return.
         status = 0
-        stdout_value: Union[str, bytes] = b""
-        stderr_value: Union[str, bytes] = b""
+        stdout_value: Union[str, bytes, None] = b""
+        stderr_value: Union[str, bytes, None] = b""
         newline = "\n" if universal_newlines else b"\n"
         try:
             if output_stream is None:
@@ -1566,7 +1624,7 @@ def communicate() -> Tuple[AnyStr, AnyStr]:
         if self.GIT_PYTHON_TRACE == "full":
             cmdstr = " ".join(redacted_command)
 
-            def as_text(stdout_value: Union[bytes, str]) -> str:
+            def as_text(stdout_value: Union[bytes, str, None]) -> str:
                 return not output_stream and safe_decode(stdout_value) or "<OUTPUT_STREAM>"
 
             # END as_text
@@ -1591,6 +1649,8 @@ def as_text(stdout_value: Union[bytes, str]) -> str:
         if isinstance(stdout_value, bytes) and stdout_as_string:  # Could also be output_stream.
             stdout_value = safe_decode(stdout_value)
 
+        # stderr is always captured through PIPE.
+        assert stderr_value is not None
         # Allow access to the command's status code.
         if with_extended_output:
             return (status, stdout_value, safe_decode(stderr_value))
@@ -1829,7 +1889,7 @@ def _parse_object_header(self, header_line: str) -> Tuple[str, str, int]:
             raise ValueError("Failed to parse header: %r" % header_line)
         return (tokens[0], tokens[1], int(tokens[2]))
 
-    def _prepare_ref(self, ref: AnyStr) -> bytes:
+    def _prepare_ref(self, ref: object) -> 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.
@@ -1856,15 +1916,15 @@ def _get_persistent_cmd(self, attr_name: str, cmd_name: str, *args: Any, **kwarg
         cmd = cast("Git.AutoInterrupt", cmd)
         return cmd
 
-    def __get_object_header(self, cmd: "Git.AutoInterrupt", ref: AnyStr) -> Tuple[str, str, int]:
+    def __get_object_header(self, cmd: "Git.AutoInterrupt", ref: Union[str, bytes]) -> Tuple[str, str, int]:
         if cmd.stdin and cmd.stdout:
             cmd.stdin.write(self._prepare_ref(ref))
             cmd.stdin.flush()
             return self._parse_object_header(cmd.stdout.readline())
         else:
             raise ValueError("cmd stdin was empty")
 
-    def get_object_header(self, ref: str) -> Tuple[str, str, int]:
+    def get_object_header(self, ref: Union[str, bytes]) -> Tuple[str, str, int]:
         """Use this method to quickly examine the type and size of the object behind the
         given ref.
 
@@ -1878,7 +1938,7 @@ def get_object_header(self, ref: str) -> Tuple[str, str, int]:
         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: str) -> Tuple[str, str, int, bytes]:
+    def get_object_data(self, ref: Union[str, bytes]) -> Tuple[str, str, int, bytes]:
         """Similar to :meth:`get_object_header`, but returns object data as well.
 
         :return:
@@ -1892,7 +1952,7 @@ def get_object_data(self, ref: str) -> Tuple[str, str, int, bytes]:
         del stream
         return (hexsha, typename, size, data)
 
-    def stream_object_data(self, ref: str) -> Tuple[str, str, int, "Git.CatFileContentStream"]:
+    def stream_object_data(self, ref: Union[str, bytes]) -> Tuple[str, str, int, "Git.CatFileContentStream"]:
         """Similar to :meth:`get_object_data`, but returns the data as a stream.
 
         :return:

git/cmd.py

@@ -1235,7 +1235,7 @@ def _read_commit_editmsg(self) -> str:
     def _commit_editmsg_filepath(self) -> str:
         return osp.join(self.repo.common_dir, "COMMIT_EDITMSG")
 
-    def _flush_stdin_and_wait(cls, proc: "Popen[bytes]", ignore_stdout: bool = False) -> bytes:
+    def _flush_stdin_and_wait(self, proc: "Popen[bytes]", ignore_stdout: bool = False) -> bytes:
         stdin_IO = proc.stdin
         if stdin_IO:
             stdin_IO.flush()

git/index/base.py

@@ -46,7 +46,7 @@
 from git.types import PathLike
 
 if TYPE_CHECKING:
-    from git.db import GitCmdObjectDB
+    from gitdb.db.base import ObjectDBR, ObjectDBW
     from git.objects.tree import TreeCacheTup
 
     from .base import IndexFile
@@ -412,7 +412,7 @@ def read_cache(
 
 
 def write_tree_from_cache(
-    entries: List[IndexEntry], odb: "GitCmdObjectDB", sl: slice, si: int = 0
+    entries: List[IndexEntry], odb: "ObjectDBW", sl: slice, si: int = 0
 ) -> Tuple[bytes, List["TreeCacheTup"]]:
     R"""Create a tree from the given sorted list of entries and put the respective
     trees into the given object database.
@@ -484,7 +484,7 @@ def _tree_entry_to_baseindexentry(tree_entry: "TreeCacheTup", stage: int) -> Bas
     return BaseIndexEntry((tree_entry[1], tree_entry[0], stage << CE_STAGESHIFT, tree_entry[2]))
 
 
-def aggressive_tree_merge(odb: "GitCmdObjectDB", tree_shas: Sequence[bytes]) -> List[BaseIndexEntry]:
+def aggressive_tree_merge(odb: "ObjectDBR", tree_shas: Sequence[bytes]) -> List[BaseIndexEntry]:
     R"""
     :return:
         List of :class:`~git.index.typ.BaseIndexEntry`\s representing the aggressive

git/index/fun.py

@@ -9,19 +9,21 @@
 from pathlib import Path
 
 from git.objects import Blob
+from git.objects.base import IndexObject
 
 from .util import pack, unpack
 
 # typing ----------------------------------------------------------------------
 
-from typing import NamedTuple, Sequence, TYPE_CHECKING, Tuple, Union, cast
+from typing import NamedTuple, Sequence, TYPE_CHECKING, Tuple, Type, TypeVar, Union, cast
 
 from git.types import PathLike
 
 if TYPE_CHECKING:
     from git.repo import Repo
 
 StageType = int
+_T_IndexEntry = TypeVar("_T_IndexEntry", bound="BaseIndexEntry")
 
 # ---------------------------------------------------------------------------------
 
@@ -104,15 +106,20 @@ class BaseIndexEntry(BaseIndexEntryHelper):
     """
 
     def __new__(
-        cls,
+        cls: Type[_T_IndexEntry],
         inp_tuple: Union[
             Tuple[int, bytes, int, PathLike],
+            Tuple[int, bytes, int, PathLike, bytes, bytes, int, int, int, int, int],
             Tuple[int, bytes, int, PathLike, bytes, bytes, int, int, int, int, int, int],
         ],
-    ) -> "BaseIndexEntry":
+    ) -> _T_IndexEntry:
         """Override ``__new__`` to allow construction from a tuple for backwards
         compatibility."""
-        return super().__new__(cls, *inp_tuple)
+        if len(inp_tuple) == 4:
+            return BaseIndexEntryHelper.__new__(cls, *inp_tuple)
+        if len(inp_tuple) == 11:
+            return BaseIndexEntryHelper.__new__(cls, *inp_tuple)
+        return BaseIndexEntryHelper.__new__(cls, *inp_tuple)
 
     def __str__(self) -> str:
         return "%o %s %i\t%s" % (self.mode, self.hexsha, self.stage, self.path)
@@ -148,7 +155,7 @@ def intent_to_add(self) -> bool:
         return (self.extended_flags & CE_EXT_INTENT_TO_ADD) > 0
 
     @classmethod
-    def from_blob(cls, blob: Blob, stage: int = 0) -> "BaseIndexEntry":
+    def from_blob(cls, blob: IndexObject, stage: int = 0) -> "BaseIndexEntry":
         """:return: Fully equipped BaseIndexEntry at the given stage"""
         return cls((blob.mode, blob.binsha, stage << CE_STAGESHIFT, blob.path))
 
@@ -192,10 +199,10 @@ def from_base(cls, base: "BaseIndexEntry") -> "IndexEntry":
             Instance of type :class:`BaseIndexEntry`.
         """
         time = pack(">LL", 0, 0)
-        return IndexEntry((base.mode, base.binsha, base.flags, base.path, time, time, 0, 0, 0, 0, 0))  # type: ignore[arg-type]
+        return IndexEntry((base.mode, base.binsha, base.flags, base.path, time, time, 0, 0, 0, 0, 0))
 
     @classmethod
-    def from_blob(cls, blob: Blob, stage: int = 0) -> "IndexEntry":
+    def from_blob(cls, blob: IndexObject, stage: int = 0) -> "IndexEntry":
         """:return: Minimal entry resembling the given blob object"""
         time = pack(">LL", 0, 0)
         return IndexEntry(
@@ -211,5 +218,5 @@ def from_blob(cls, blob: Blob, stage: int = 0) -> "IndexEntry":
                 0,
                 0,
                 blob.size,
-            )  # type: ignore[arg-type]
+            )
         )

git/index/typ.py

@@ -18,7 +18,7 @@
 
 from typing import Any, TYPE_CHECKING, Union
 
-from git.types import AnyGitObject, GitObjectTypeString, PathLike
+from git.types import AnyGitObject, GitObjectTypeString, PathLike, SupportsWrite
 
 if TYPE_CHECKING:
     from gitdb.base import OStream
@@ -200,7 +200,7 @@ def data_stream(self) -> "OStream":
         """
         return self.repo.odb.stream(self.binsha)
 
-    def stream_data(self, ostream: "OStream") -> "Object":
+    def stream_data(self, ostream: SupportsWrite[bytes]) -> "Object":
         """Write our data directly to the given output stream.
 
         :param ostream:

git/objects/base.py

@@ -502,7 +502,7 @@ def _interpret_trailers(
     ) -> str:
         message_bytes = message if isinstance(message, bytes) else message.encode(encoding, errors="strict")
         cmd = [repo.git.GIT_PYTHON_GIT_EXECUTABLE, "interpret-trailers", *trailer_args]
-        proc: Git.AutoInterrupt = repo.git.execute(  # type: ignore[call-overload]
+        proc: Git.AutoInterrupt = repo.git.execute(
             cmd,
             as_process=True,
             istream=PIPE,