Fix mypy type errors and add GitClient.close() method

dulwich/__init__.py

@@ -23,7 +23,7 @@
 """Python implementation of the Git file formats and protocols."""
 
 from collections.abc import Callable
-from typing import Any, ParamSpec, TypeVar
+from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar
 
 __version__ = (0, 25, 0)
 
@@ -33,51 +33,61 @@ P = ParamSpec("P")
 R = TypeVar("R")
 F = TypeVar("F", bound=Callable[..., Any])
 
-try:
-    from dissolve import replace_me as replace_me
-except ImportError:
-    # if dissolve is not installed, then just provide a basic implementation
-    # of its replace_me decorator
+if TYPE_CHECKING:
+    # For type checking, always use our typed signature
     def replace_me(
         since: tuple[int, ...] | str | None = None,
         remove_in: tuple[int, ...] | str | None = None,
-    ) -> Callable[[F], F]:
-        """Decorator to mark functions as deprecated.
+    ) -> Callable[[Callable[P, R]], Callable[P, R]]:
+        """Decorator to mark functions as deprecated."""
+        ...
 
-        Args:
-            since: Version when the function was deprecated
-            remove_in: Version when the function will be removed
+else:
+    try:
+        from dissolve import replace_me as replace_me
+    except ImportError:
+        # if dissolve is not installed, then just provide a basic implementation
+        # of its replace_me decorator
+        def replace_me(
+            since: tuple[int, ...] | str | None = None,
+            remove_in: tuple[int, ...] | str | None = None,
+        ) -> Callable[[Callable[P, R]], Callable[P, R]]:
+            """Decorator to mark functions as deprecated.
 
-        Returns:
-            Decorator function
-        """
+            Args:
+                since: Version when the function was deprecated
+                remove_in: Version when the function will be removed
 
-        def decorator(func: Callable[P, R]) -> Callable[P, R]:
-            import functools
-            import warnings
+            Returns:
+                Decorator function
+            """
 
-            m = f"{func.__name__} is deprecated"
-            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
+            def decorator(func: Callable[P, R]) -> Callable[P, R]:
+                import functools
+                import warnings
 
-            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"
+                m = f"{func.__name__} is deprecated"
+                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
 
-            @functools.wraps(func)
-            def _wrapped_func(*args: P.args, **kwargs: P.kwargs) -> R:
-                warnings.warn(
-                    m,
-                    DeprecationWarning,
-                    stacklevel=2,
-                )
-                return func(*args, **kwargs)
+                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"
 
-            return _wrapped_func
+                @functools.wraps(func)
+                def _wrapped_func(*args: P.args, **kwargs: P.kwargs) -> R:
+                    warnings.warn(
+                        m,
+                        DeprecationWarning,
+                        stacklevel=2,
+                    )
+                    return func(*args, **kwargs)
 
-        return decorator  # type: ignore[return-value]
+                return _wrapped_func
+
+            return decorator

dulwich/aiohttp/server.py

@@ -30,6 +30,7 @@ from aiohttp import web
 
 from .. import log_utils
 from ..errors import HangupException
+from ..objects import ObjectID
 from ..protocol import ReceivableProtocol
 from ..repo import Repo
 from ..server import (
@@ -85,7 +86,9 @@ async def get_loose_object(request: web.Request) -> web.Response:
       request: aiohttp request object
     Returns: Response with the loose object data
     """
-    sha = (request.match_info["dir"] + request.match_info["file"]).encode("ascii")
+    sha = ObjectID(
+        (request.match_info["dir"] + request.match_info["file"]).encode("ascii")
+    )
     logger.info("Sending loose object %s", sha)
     object_store = request.app[REPO_KEY].object_store
     if not object_store.contains_loose(sha):

dulwich/bundle.py

@@ -29,6 +29,7 @@ __all__ = [
     "write_bundle",
 ]
 
+import types
 from collections.abc import Callable, Iterator, Sequence
 from typing import (
     TYPE_CHECKING,
@@ -60,6 +61,10 @@ class PackDataLike(Protocol):
         """Iterate over unpacked objects in the pack."""
         ...
 
+    def close(self) -> None:
+        """Close any open resources."""
+        ...
+
 
 if TYPE_CHECKING:
     from .object_store import BaseObjectStore
@@ -111,7 +116,12 @@ class Bundle:
         """Enter context manager."""
         return self
 
-    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: types.TracebackType | None,
+    ) -> None:
         """Exit context manager and close bundle."""
         self.close()
 
@@ -119,6 +129,7 @@ class Bundle:
         """Warn if bundle was not explicitly closed."""
         if self.pack_data is not None:
             import warnings
+
             warnings.warn(
                 f"Bundle {self!r} was not explicitly closed. "
                 "Please use bundle.close() or a context manager.",
@@ -359,7 +370,6 @@ def create_bundle_from_repo(
 
         def close(self) -> None:
             """Close pack data (no-op for in-memory pack data)."""
-            pass
 
     pack_data = _BundlePackData(pack_count, pack_objects, repo.object_format)
 

dulwich/client.py

@@ -1020,6 +1020,14 @@ class GitClient:
             self._fetch_capabilities.add(CAPABILITY_INCLUDE_TAG)
         self.protocol_version = 0  # will be overridden later
 
+    def close(self) -> None:
+        """Close the client and release any resources.
+
+        Default implementation does nothing as most clients don't maintain
+        persistent connections. Subclasses that hold resources should override
+        this method to properly clean them up.
+        """
+
     def get_url(self, path: str) -> str:
         """Retrieves full url to given path.
 
@@ -4598,9 +4606,9 @@ class Urllib3HttpGitClient(AbstractHttpGitClient):
             if resp.status != 200:
                 raise GitProtocolError(f"unexpected http resp {resp.status} for {url}")
 
-        resp.content_type = resp.headers.get("Content-Type")  # type: ignore[attr-defined]
+        resp.content_type = resp.headers.get("Content-Type")  # type: ignore[union-attr]
         resp_url = resp.geturl()
-        resp.redirect_location = resp_url if resp_url != url else ""  # type: ignore[attr-defined]
+        resp.redirect_location = resp_url if resp_url != url else ""  # type: ignore[union-attr]
         return resp, _wrap_urllib3_exceptions(resp.read)  # type: ignore[return-value]
 
 

dulwich/object_store.py

@@ -1676,7 +1676,7 @@ class DiskObjectStore(PackBasedObjectStore):
                 idx_name = os.path.splitext(name)[0] + ".idx"
                 if idx_name in pack_dir_contents:
                     # Extract just the hash (remove "pack-" prefix and ".pack" suffix)
-                    pack_hash = name[len("pack-"): -len(".pack")]
+                    pack_hash = name[len("pack-") : -len(".pack")]
                     pack_files.add(pack_hash)
 
         # Open newly appeared pack files
@@ -1942,7 +1942,7 @@ class DiskObjectStore(PackBasedObjectStore):
         )
         final_pack.check_length_and_checksum()
         # Extract just the hash from pack_base_name (/path/to/pack-HASH -> HASH)
-        pack_hash = os.path.basename(pack_base_name)[len("pack-"):]
+        pack_hash = os.path.basename(pack_base_name)[len("pack-") :]
         self._add_cached_pack(pack_hash, final_pack)
         return final_pack
 

dulwich/pack.py

@@ -4085,10 +4085,10 @@ class Pack:
         """Close the pack file and index."""
         if self._data is not None:
             self._data.close()
-            self._data = None  # type: ignore
+            self._data = None
         if self._idx is not None:
             self._idx.close()
-            self._idx = None  # type: ignore
+            self._idx = None
 
     def __del__(self) -> None:
         """Ensure pack file is closed when Pack is garbage collected."""

tests/test_pack.py

@@ -478,7 +478,9 @@ class TestPackData(PackTests):
             self.datadir, "pack-{}.pack".format(pack1_sha.decode("ascii"))
         )
         with open(path, "rb") as f:
-            pack_data = PackData.from_file(f, DEFAULT_OBJECT_FORMAT, os.path.getsize(path))
+            pack_data = PackData.from_file(
+                f, DEFAULT_OBJECT_FORMAT, os.path.getsize(path)
+            )
             pack_data.close()
 
     def test_pack_len(self) -> None: