Improve type hints

Author: alorence

.claude/settings.local.json

@@ -0,0 +1,9 @@
+{
+  "permissions": {
+    "allow": [
+      "Bash(uv run mypy:*)",
+      "Bash(uv run ty check:*)",
+      "Bash(uv run pytest:*)"
+    ]
+  }
+}

modernrpc/apps.py

@@ -9,4 +9,4 @@ class ModernRpcConfig(AppConfig):
     name = "modernrpc"
     verbose_name = "Django Modern RPC"
 
-    def ready(self): ...
+    def ready(self) -> None: ...

modernrpc/compat.py

@@ -5,6 +5,13 @@
 import django
 from django.views.decorators.csrf import csrf_exempt
 
+try:
+    # types.NoneType is available only with Python 3.10+
+    from types import NoneType
+except ImportError:
+    NoneType = type(None)  # type: ignore[misc]  # ty: ignore[unused-type-ignore-comment
+
+
 if django.VERSION >= (5, 0):
     # Django updated its decorators to support wrapping asynchronous method in release 5.0
     # REF: https://docs.djangoproject.com/en/5.2/releases/5.0/#decorators
@@ -21,26 +28,26 @@ def async_csrf_exempt(view):
 
 if sys.version_info >= (3, 14):
 
-    def is_union_type(_type: type):
+    def is_union_type(_type: typing.Any) -> bool:
         return isinstance(_type, types.UnionType)
 
 elif sys.version_info >= (3, 10):
 
-    def is_union_type(_type: type):
+    def is_union_type(_type: typing.Any) -> bool:
         return typing.get_origin(_type) in (types.UnionType, typing.Union)
 
 else:
 
-    def is_union_type(_type: type):
+    def is_union_type(_type: typing.Any) -> bool:
         return typing.get_origin(_type) is typing.Union
 
 
 if sys.version_info >= (3, 14):
 
-    def union_str_repr(obj):
+    def union_str_repr(obj: typing.Any) -> str:
         return str(obj)
 
 else:
 
-    def union_str_repr(obj: type):
+    def union_str_repr(obj: typing.Any) -> str:
         return " | ".join(t.__name__ for t in typing.get_args(obj))

modernrpc/core.py

@@ -4,7 +4,7 @@
 import logging
 from collections import OrderedDict, defaultdict
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any, Callable, Iterable
+from typing import TYPE_CHECKING, Any, Iterable
 
 from asgiref.sync import async_to_sync, iscoroutinefunction, sync_to_async
 from django.utils.functional import cached_property
@@ -25,7 +25,7 @@
 
     from modernrpc.handler import RpcHandler
     from modernrpc.server import RpcServer
-    from modernrpc.types import AuthPredicateType
+    from modernrpc.types import AuthPredicateType, FuncOrCoro
 
 logger = logging.getLogger(__name__)
 
@@ -57,7 +57,7 @@ class RpcRequestContext:
 class ProcedureArgDocs:
     docstring: str = ""
     doc_type: str = ""
-    type_hint: type | None = None
+    type_hint: Any = None
 
     @property
     def type_hint_as_str(self) -> str:
@@ -79,7 +79,7 @@ class ProcedureWrapper:
 
     def __init__(
         self,
-        func_or_coro: Callable,
+        func_or_coro: FuncOrCoro,
         name: str | None = None,
         protocol: Protocol = Protocol.ALL,
         auth: AuthPredicateType = NOT_SET,
@@ -89,7 +89,8 @@ def __init__(
         self.func_or_coro = func_or_coro
 
         # @decorator parameters
-        self.name = name or func_or_coro.__name__
+        func_name: str = getattr(func_or_coro, "__name__", None) or str(func_or_coro)
+        self.name: str = name or func_name
         self.protocol = protocol
         self.context_target = context_target
 

modernrpc/helpers.py

@@ -64,9 +64,9 @@ def first(seq: Iterable, default=NOT_SET) -> Any:
     try:
         return next(iter(seq))
     except StopIteration:
-        if default is not NOT_SET:
-            return default
-        raise IndexError("No element found in iterable") from None
+        if default is NOT_SET:
+            raise IndexError("No element found in iterable") from None
+        return default
 
 
 def first_true(iterable: Iterable[Any], default: Any = None, pred: Callable[[Any], bool] | None = None) -> Any:

modernrpc/introspection.py

@@ -8,15 +8,18 @@
 import inspect
 import re
 import typing
-from typing import Callable
+from typing import TYPE_CHECKING
 
 from django.utils.functional import cached_property
 
+if TYPE_CHECKING:
+    from modernrpc.types import FuncOrCoro
+
 
 class Introspector:
     """Helper class to extract the signature of a callable and the type hint of its arguments & returns"""
 
-    def __init__(self, function: Callable):
+    def __init__(self, function: FuncOrCoro):
         self.func = function
 
     @cached_property
@@ -45,7 +48,7 @@ class DocstringParser:
     PARAM_TYPE_REXP = re.compile(r"^[:@]type (\w+): ?(.*)$", flags=re.MULTILINE)
     RETURN_TYPE_REXP = re.compile(r"^[:@]rtype ?: ?(.*)$", flags=re.MULTILINE)
 
-    def __init__(self, function: Callable):
+    def __init__(self, function: FuncOrCoro):
         self.func = function
 
     @staticmethod

modernrpc/jsonrpc/backends/marshalling.py

@@ -2,19 +2,14 @@
 # PEP 604: use of typeA | typeB is available since Python 3.10, enable it for older versions
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Iterable, overload
+from typing import TYPE_CHECKING, cast, overload
 
+from modernrpc.compat import NoneType
 from modernrpc.constants import NOT_SET
 from modernrpc.exceptions import RPCInvalidRequest
 from modernrpc.jsonrpc.handler import JsonRpcRequest
 from modernrpc.types import DictStrAny, RpcErrorResult
 
-try:
-    # types.NoneType is available only with Python 3.10+
-    from types import NoneType
-except ImportError:
-    NoneType = type(None)  # type: ignore[misc]
-
 if TYPE_CHECKING:
     from modernrpc.jsonrpc.handler import JsonRpcResult
 
@@ -75,13 +70,13 @@ class Marshaller:
     def result_to_dict(self, result: JsonRpcResult) -> DictStrAny | None: ...
 
     @overload
-    def result_to_dict(self, result: Iterable[JsonRpcResult]) -> list[DictStrAny | None]: ...
+    def result_to_dict(self, result: list[JsonRpcResult]) -> list[DictStrAny | None]: ...
 
     def result_to_dict(
-        self, result: JsonRpcResult | Iterable[JsonRpcResult]
+        self, result: JsonRpcResult | list[JsonRpcResult]
     ) -> DictStrAny | None | list[DictStrAny | None]:
-        if isinstance(result, Iterable):
-            return [self.result_to_dict(r) for r in result]
+        if isinstance(result, list):
+            return [self.result_to_dict(cast("JsonRpcResult", r)) for r in result]
 
         if result.request.is_notification:
             return None

modernrpc/jsonrpc/handler.py

@@ -6,7 +6,7 @@
 import logging
 from dataclasses import dataclass, field
 from http import HTTPStatus
-from typing import TYPE_CHECKING, Union
+from typing import TYPE_CHECKING, Union, cast
 
 from django.utils.module_loading import import_string
 
@@ -84,9 +84,9 @@ def process_request(self, request_body: str, context: RpcRequestContext) -> str
             exc = context.server.on_error(exc, context)
             return self.serializer.dumps(self.build_error_result(fake_request, exc.code, exc.message))
 
-        # Parsed request is an Iterable, we should handle it as a batch request
+        # Parsed request is a list, we should handle it as a batch request
         if isinstance(parsed_request, list):
-            return self.process_batch_request(parsed_request, context)
+            return self.process_batch_request(cast("list[JsonRpcRequest]", parsed_request), context)
 
         # By default, handle a standard single request
         result = self.process_single_request(parsed_request, context)
@@ -117,9 +117,9 @@ async def aprocess_request(self, request_body: str, context: RpcRequestContext)
             exc = context.server.on_error(exc, context)
             return self.serializer.dumps(self.build_error_result(fake_request, exc.code, exc.message))
 
-        # Parsed request is an Iterable, we should handle it as a batch request
+        # Parsed request is a list, we should handle it as a batch request
         if isinstance(parsed_request, list):
-            return await self.aprocess_batch_request(parsed_request, context)
+            return await self.aprocess_batch_request(cast("list[JsonRpcRequest]", parsed_request), context)
 
         # By default, handle a standard single request
         result = await self.aprocess_single_request(parsed_request, context)

modernrpc/server.py

@@ -25,22 +25,26 @@
     from django.shortcuts import SupportsGetAbsoluteUrl
 
     from modernrpc.handler import RpcHandler
-    from modernrpc.types import AuthPredicateType
-
+    from modernrpc.types import AuthPredicateType, FuncOrCoro
 
 logger = logging.getLogger(__name__)
 
 RpcErrorHandler = Callable[[BaseException, RpcRequestContext], None]
 
 
 class RegistryMixin:
+    """
+    Common logic for both RpcNamespace and RpcServer classes.
+    Provide methods to register RPC procedures into an internal registry.
+    """
+
     def __init__(self, auth: AuthPredicateType = NOT_SET) -> None:
         self._registry: dict[str, ProcedureWrapper] = {}
         self.auth = auth
 
     def register_procedure(
         self,
-        procedure: Callable | None = None,
+        procedure: FuncOrCoro | None = None,
         name: str | None = None,
         protocol: Protocol = Protocol.ALL,
         auth: AuthPredicateType = NOT_SET,
@@ -61,7 +65,7 @@ def register_procedure(
         :raises ValueError: If a procedure can't be registered
         """
 
-        def decorated(func: Callable) -> Callable:
+        def decorated(func: FuncOrCoro) -> FuncOrCoro:
             if name and name.startswith("rpc."):
                 raise ValueError(
                     'According to JSON-RPC specs, method names starting with "rpc." are reserved for system extensions '
@@ -92,10 +96,13 @@ def procedures(self) -> dict[str, ProcedureWrapper]:
         return self._registry
 
 
-class RpcNamespace(RegistryMixin): ...
+class RpcNamespace(RegistryMixin):
+    """Registry for RPC procedures belonging to a given namespace."""
 
 
 class RpcServer(RegistryMixin):
+    """Base class to store all remote procedures for a given entry point"""
+
     def __init__(
         self,
         register_system_procedures: bool = True,
@@ -122,6 +129,7 @@ def __init__(
         self.default_encoding = default_encoding
 
     def register_namespace(self, namespace: RpcNamespace, name: str | None = None) -> None:
+        """Register all procedures from given namespace into the top-level server."""
         if name:
             prefix = name + "."
             logger.debug(
@@ -157,6 +165,7 @@ def get_procedure_wrapper(self, name: str, protocol: Protocol) -> ProcedureWrapp
         raise RPCMethodNotFound(name) from None
 
     def get_request_handler(self, request: HttpRequest) -> RpcHandler | None:
+        """Return the first handler that can handle the given request, or None if no handler can handle it."""
         handler_klass = first_true(self.handler_klasses, pred=lambda cls: cls.can_handle(request))
         try:
             return handler_klass()
@@ -217,6 +226,7 @@ def check_request(self, request: HttpRequest) -> HttpResponse | None:
 
     @staticmethod
     def build_response(handler: RpcHandler, result_data: str | tuple[int, str]) -> HttpResponse:
+        """Build an HttpResponse instance from the given handler and result data."""
         if isinstance(result_data, tuple) and len(result_data) == 2:
             status, result_data = result_data
         else:

modernrpc/types.py

@@ -42,3 +42,5 @@ class RpcErrorResult(RpcResult[RequestType]):
 NotSetType = object
 AuthPredicate = Callable[[RpcRequest], bool]
 AuthPredicateType = Union[NotSetType, AuthPredicate, Sequence[AuthPredicate]]
+
+FuncOrCoro = Callable[..., Any]