Typing what we can

pyproject.toml

@@ -9,7 +9,8 @@ classifiers = [
     "License :: OSI Approved :: MIT License",
     "Topic :: Internet :: WWW/HTTP",
     "Topic :: Software Development :: Libraries",
-    "Topic :: System :: Networking"
+    "Topic :: System :: Networking",
+    "Typing :: Typed",
 ]
 version = "0.8.1"
 readme = "README.rst"
@@ -32,12 +33,22 @@ Changelog = "https://mardiros.github.io/aioxmlrpc/user/changelog.html"
 
 [tool.pytest.ini_options]
 asyncio_mode = "auto"
-asyncio_default_fixture_loop_scope="function"
+asyncio_default_fixture_loop_scope = "function"
+
+[[tool.mypy.overrides]]
+disallow_any_generics = true
+disallow_untyped_defs = true
+module = "aioxmlrpc.*"
 
 [tool.pyright]
 ignore = ["examples"]
 include = ["src", "tests"]
-typeCheckingMode = "basic"
+reportPrivateUsage = false
+reportUnknownMemberType = false
+reportUnknownParameterType = false
+reportUnknownVariableType = false
+reportShadowedImports = false
+typeCheckingMode = "strict"
 venvPath = ".venv"
 
 [dependency-groups]

src/aioxmlrpc/client.py

@@ -8,12 +8,22 @@
 
 import asyncio
 import logging
+from typing import (
+    Any,
+    Awaitable,
+    Callable,
+    Optional,
+    cast,
+)
 from xmlrpc import client as xmlrpc
 
 import httpx
 
 __ALL__ = ["ServerProxy", "Fault", "ProtocolError"]
 
+RPCResult = Any
+RPCParameters = Any
+
 # you don't have to import xmlrpc.client from your code
 Fault = xmlrpc.Fault
 ProtocolError = xmlrpc.ProtocolError
@@ -24,14 +34,16 @@
 class _Method:
     # some magic to bind an XML-RPC method to an RPC server.
     # supports "nested" methods (e.g. examples.getStateName)
-    def __init__(self, send, name):
+    def __init__(
+        self, send: Callable[[str, RPCParameters], Awaitable[RPCResult]], name: str
+    ) -> None:
         self.__send = send
         self.__name = name
 
-    def __getattr__(self, name):
+    def __getattr__(self, name: str) -> "_Method":
         return _Method(self.__send, "%s.%s" % (self.__name, name))
 
-    async def __call__(self, *args):
+    async def __call__(self, *args: RPCParameters) -> RPCResult:
         ret = await self.__send(self.__name, args)
         return ret
 
@@ -43,22 +55,28 @@ class AioTransport(xmlrpc.Transport):
 
     def __init__(
         self,
-        session,
-        use_https,
+        session: httpx.AsyncClient,
+        use_https: bool,
         *,
-        use_datetime=False,
-        use_builtin_types=False,
-        auth=None,
-        timeout=None,
+        use_datetime: bool = False,
+        use_builtin_types: bool = False,
+        auth: Optional[httpx._types.AuthTypes] = None,
+        timeout: Optional[httpx._types.TimeoutTypes] = None,
     ):
         super().__init__(use_datetime, use_builtin_types)
         self.use_https = use_https
         self._session = session
 
-        self.auth = auth
+        self.auth = auth or httpx.USE_CLIENT_DEFAULT
         self.timeout = timeout
 
-    async def request(self, host, handler, request_body, verbose=False):
+    async def request(  # type: ignore
+        self,
+        host: str,
+        handler: str,
+        request_body: dict[str, Any],
+        verbose: bool = False,
+    ) -> RPCResult:
         """
         Send the XML-RPC request, return the response.
         This method is a coroutine.
@@ -78,7 +96,9 @@ async def request(self, host, handler, request_body, verbose=False):
                     url,
                     response.status_code,
                     body,
-                    response.headers,
+                    # response.headers is a case insensitive dict from httpx,
+                    # the ProtocolError is typed as simple dict
+                    cast(dict[str, str], response.headers),
                 )
         except asyncio.CancelledError:
             raise
@@ -88,15 +108,18 @@ async def request(self, host, handler, request_body, verbose=False):
             log.error("Unexpected error", exc_info=True)
             if response is not None:
                 errcode = response.status_code
-                headers = response.headers
+                headers = cast(dict[str, str], response.headers)
             else:
                 errcode = 0
                 headers = {}
 
             raise ProtocolError(url, errcode, str(exc), headers)
         return self.parse_response(body)
 
-    def parse_response(self, body):
+    def parse_response(  # type: ignore
+        self,
+        body: str,
+    ) -> RPCResult:
         """
         Parse the xmlrpc response.
         """
@@ -105,13 +128,13 @@ def parse_response(self, body):
         p.close()
         return u.close()
 
-    def _build_url(self, host, handler):
+    def _build_url(self, host: str, handler: str) -> str:
         """
         Build a url for our request based on the host, handler and use_http
         property
         """
         scheme = "https" if self.use_https else "http"
-        return "%s://%s%s" % (scheme, host, handler)
+        return f"{scheme}://{host}{handler}"
 
 
 class ServerProxy(xmlrpc.ServerProxy):
@@ -121,19 +144,19 @@ class ServerProxy(xmlrpc.ServerProxy):
 
     def __init__(
         self,
-        uri,
-        encoding=None,
-        verbose=False,
-        allow_none=False,
-        use_datetime=False,
-        use_builtin_types=False,
-        auth=None,
+        uri: str,
+        encoding: Optional[str] = None,
+        verbose: bool = False,
+        allow_none: bool = False,
+        use_datetime: bool = False,
+        use_builtin_types: bool = False,
+        auth: Optional[httpx._types.AuthTypes] = None,
         *,
-        headers=None,
-        context=None,
-        timeout=5.0,
-        session=None,
-    ):
+        headers: Optional[dict[str, Any]] = None,
+        context: Optional[httpx._types.VerifyTypes] = None,
+        timeout: httpx._types.TimeoutTypes = 5.0,
+        session: Optional[httpx.AsyncClient] = None,
+    ) -> None:
         if not headers:
             headers = {
                 "User-Agent": "python/aioxmlrpc",
@@ -162,20 +185,24 @@ def __init__(
             use_builtin_types,
         )
 
-    async def __request(self, methodname, params):
+    async def __request(  # type: ignore
+        self,
+        methodname: str,
+        params: RPCParameters,
+    ) -> RPCResult:
         # call a method on the remote server
         request = xmlrpc.dumps(
             params, methodname, encoding=self.__encoding, allow_none=self.__allow_none
         ).encode(self.__encoding)
 
-        response = await self.__transport.request(
+        response = await self.__transport.request(  # type: ignore
             self.__host, self.__handler, request, verbose=self.__verbose
         )
 
-        if len(response) == 1:
+        if len(response) == 1:  # type: ignore
             response = response[0]
 
         return response
 
-    def __getattr__(self, name):
+    def __getattr__(self, name: str) -> _Method:  # type: ignore
         return _Method(self.__request, name)