Add more typing (#1745)

dulwich/__init__.py

@@ -24,7 +24,7 @@
 """Python implementation of the Git file formats and protocols."""
 
 import sys
-from typing import Callable, Optional, TypeVar
+from typing import Optional, TypeVar, Union
 
 if sys.version_info >= (3, 10):
     from typing import ParamSpec
@@ -37,6 +37,7 @@ __all__ = ["__version__", "replace_me"]
 
 P = ParamSpec("P")
 R = TypeVar("R")
+F = TypeVar("F")
 
 try:
     from dissolve import replace_me
@@ -44,24 +45,28 @@ except ImportError:
     # if dissolve is not installed, then just provide a basic implementation
     # of its replace_me decorator
     def replace_me(
-        since: Optional[str] = None, remove_in: Optional[str] = None
-    ) -> Callable[[Callable[P, R]], Callable[P, R]]:
-        def decorator(func: Callable[P, R]) -> Callable[P, R]:
+        since: Optional[Union[str, tuple[int, ...]]] = None,
+        remove_in: Optional[Union[str, tuple[int, ...]]] = None,
+    ):
+        def decorator(func):
             import functools
             import warnings
 
             m = f"{func.__name__} is deprecated"
-            if since is not None and remove_in is not None:
-                m += f" since {since} and will be removed in {remove_in}"
-            elif since is not None:
-                m += f" since {since}"
-            elif remove_in is not None:
-                m += f" and will be removed in {remove_in}"
+            since_str = str(since) if since is not None else None
+            remove_in_str = str(remove_in) if remove_in is not None else None
+
+            if since_str is not None and remove_in_str is not None:
+                m += f" since {since_str} and will be removed in {remove_in_str}"
+            elif since_str is not None:
+                m += f" since {since_str}"
+            elif remove_in_str is not None:
+                m += f" and will be removed in {remove_in_str}"
             else:
                 m += " and will be removed in a future version"
 
             @functools.wraps(func)
-            def _wrapped_func(*args: P.args, **kwargs: P.kwargs) -> R:
+            def _wrapped_func(*args, **kwargs):
                 warnings.warn(
                     m,
                     DeprecationWarning,

dulwich/index.py

@@ -46,6 +46,7 @@ if TYPE_CHECKING:
     from .diff_tree import TreeChange
     from .file import _GitFile
     from .line_ending import BlobNormalizer
+    from .object_store import BaseObjectStore
     from .repo import Repo
 
 from .file import GitFile
@@ -1324,7 +1325,7 @@ def _normalize_path_element_hfs(element: bytes) -> bytes:
     return normalized.lower().encode("utf-8", errors="strict")
 
 
-def get_path_element_normalizer(config) -> Callable[[bytes], bytes]:
+def get_path_element_normalizer(config: "Config") -> Callable[[bytes], bytes]:
     """Get the appropriate path element normalization function based on config.
 
     Args:
@@ -1645,7 +1646,7 @@ def _remove_empty_parents(path: bytes, stop_at: bytes) -> None:
 
 
 def _check_symlink_matches(
-    full_path: bytes, repo_object_store, entry_sha: bytes
+    full_path: bytes, repo_object_store: "BaseObjectStore", entry_sha: bytes
 ) -> bool:
     """Check if symlink target matches expected target.
 
@@ -1669,7 +1670,7 @@ def _check_symlink_matches(
 
 
 def _check_file_matches(
-    repo_object_store,
+    repo_object_store: "BaseObjectStore",
     full_path: bytes,
     entry_sha: bytes,
     entry_mode: int,
@@ -1724,6 +1725,7 @@ def _check_file_matches(
             current_content = f.read()
             expected_content = blob_obj.as_raw_string()
             if blob_normalizer and tree_path is not None:
+                assert isinstance(blob_obj, Blob)
                 normalized_blob = blob_normalizer.checkout_normalize(
                     blob_obj, tree_path
                 )
@@ -1733,7 +1735,14 @@ def _check_file_matches(
         return False
 
 
-def _transition_to_submodule(repo, path, full_path, current_stat, entry, index):
+def _transition_to_submodule(
+    repo: "Repo",
+    path: bytes,
+    full_path: bytes,
+    current_stat: Optional[os.stat_result],
+    entry: IndexEntry,
+    index: Index,
+) -> None:
     """Transition any type to submodule."""
     from .submodule import ensure_submodule_placeholder
 
@@ -1751,17 +1760,17 @@ def _transition_to_submodule(repo, path, full_path, current_stat, entry, index):
 
 
 def _transition_to_file(
-    object_store,
-    path,
-    full_path,
-    current_stat,
-    entry,
-    index,
-    honor_filemode,
-    symlink_fn,
-    blob_normalizer,
-    tree_encoding="utf-8",
-):
+    object_store: "BaseObjectStore",
+    path: bytes,
+    full_path: bytes,
+    current_stat: Optional[os.stat_result],
+    entry: IndexEntry,
+    index: Index,
+    honor_filemode: bool,
+    symlink_fn: Optional[Callable[[bytes, bytes], None]],
+    blob_normalizer: Optional["BlobNormalizer"],
+    tree_encoding: str = "utf-8",
+) -> None:
     """Transition any type to regular file or symlink."""
     # Check if we need to update
     if (
@@ -1794,6 +1803,7 @@ def _transition_to_file(
 
     if not needs_update:
         # Just update index - current_stat should always be valid here since we're not updating
+        assert current_stat is not None
         index[path] = index_entry_from_stat(current_stat, entry.sha)
         return
 
@@ -1840,7 +1850,13 @@ def _transition_to_file(
     index[path] = index_entry_from_stat(st, entry.sha)
 
 
-def _transition_to_absent(repo, path, full_path, current_stat, index):
+def _transition_to_absent(
+    repo: "Repo",
+    path: bytes,
+    full_path: bytes,
+    current_stat: Optional[os.stat_result],
+    index: Index,
+) -> None:
     """Remove any type of entry."""
     if current_stat is None:
         return

dulwich/worktree.py

@@ -31,7 +31,7 @@ import sys
 import tempfile
 import time
 import warnings
-from collections.abc import Iterable
+from collections.abc import Iterable, Iterator
 from contextlib import contextmanager
 from pathlib import Path
 
@@ -219,7 +219,7 @@ class WorkTreeContainer:
         """
         unlock_worktree(self._repo, path)
 
-    def __iter__(self):
+    def __iter__(self) -> Iterator[WorkTreeInfo]:
         """Iterate over all worktrees."""
         yield from self.list()
 
@@ -374,17 +374,17 @@ class WorkTree:
         message: bytes | None = None,
         committer: bytes | None = None,
         author: bytes | None = None,
-        commit_timestamp=None,
-        commit_timezone=None,
-        author_timestamp=None,
-        author_timezone=None,
+        commit_timestamp: float | None = None,
+        commit_timezone: int | None = None,
+        author_timestamp: float | None = None,
+        author_timezone: int | None = None,
         tree: ObjectID | None = None,
         encoding: bytes | None = None,
         ref: Ref | None = b"HEAD",
         merge_heads: list[ObjectID] | None = None,
         no_verify: bool = False,
         sign: bool = False,
-    ):
+    ) -> ObjectID:
         """Create a new commit.
 
         If not specified, committer and author default to
@@ -569,7 +569,7 @@ class WorkTree:
 
         return c.id
 
-    def reset_index(self, tree: bytes | None = None):
+    def reset_index(self, tree: bytes | None = None) -> None:
         """Reset the index back to a specific tree.
 
         Args:
@@ -1157,7 +1157,7 @@ def move_worktree(
 
 
 @contextmanager
-def temporary_worktree(repo, prefix="tmp-worktree-"):
+def temporary_worktree(repo: Repo, prefix: str = "tmp-worktree-") -> Iterator[Repo]:
     """Create a temporary worktree that is automatically cleaned up.
 
     Args: