Correct public API typing and add portable runtime checks

Several public annotations rejected supported inputs or lost the relationship
between input options and return types. Describe Git.execute process, text,
bytes, and extended-output results with overloads, accept stdin file
descriptors, and account for absent stdout. Correct remote-removal, object,
database, blame, and index-entry types, preserve entry subclasses and the
supported tuple shapes, and accept streams with only the required read or
write methods. Normalize absent previous stderr before appending process
errors in AutoInterrupt.wait.

Add runtime and static regressions for these interfaces, include them in
mypy and basedpyright, make required imports explicit, and reduce the
basedpyright baseline to the remaining diagnostics. Keep mock available to
typecheck the Python 3.7 import branches.

Make the output-type regression emit its payload without a newline. Its
original print call produced CRLF on Windows; Git.execute strips the final
LF as documented, leaving a carriage return that failed the test in all
11 Windows jobs. Omitting the newline keeps the same text, bytes, and tuple
assertions independent of platform newline translation.

Validation on macOS with Python 3.12.14: all four test/test_typing.py tests
pass. Forcing the child stdout to translate newlines to CRLF reproduces the
old failure and passes with the corrected command. Mypy passes for 46
source files; basedpyright --warnings reports no errors or warnings; Ruff
lint and format checks for test/test_typing.py and git diff --check pass.
Native Windows validation is delegated to the PR CI matrix.

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,