Merge pull request #32 from mardiros/features/mypy

Features/mypy

.github/workflows/tests.yml

@@ -1,33 +1,22 @@
-# This is a basic workflow to help you get started with Actions
-
 name: tests
 
-# Controls when the workflow will run
 on:
-  # Triggers the workflow on push or pull request events but only for the main branch
   push:
     branches: [ main ]
   pull_request:
     branches: [ main ]
 
-  # Allows you to run the tests suite manually from the Actions tab
   workflow_dispatch:
 
-  # Allows you to run the tests suite before building release artifacts
   workflow_call:
 
-# A workflow run is made up of one or more jobs that can run sequentially or in parallel
 jobs:
-  # This workflow contains a single job called "build"
-  CI:
-    # The type of runner that the job will run on
+  tests:
     runs-on: ubuntu-latest
     strategy:
       matrix:
         python-version: ["3.9", "3.10", "3.11", "3.12"]
-    # Steps represent a sequence of tasks that will be executed as part of the job
     steps:
-      # Checks-out your repository under $GITHUB_WORKSPACE, so your job can access it
       - uses: actions/checkout@v4
       - uses: chartboost/ruff-action@v1
 
@@ -41,9 +30,9 @@ jobs:
       - name: Install the project
         run: uv sync --group dev
 
-      # - name: Check types
-      #   run: |
-      #     uv run mypy src/aioxmlrpc/
+      - name: Check types
+        run: |
+          uv run mypy src/aioxmlrpc/
 
       - name: Run tests
         run: |
@@ -66,5 +55,4 @@ jobs:
         uses: codecov/codecov-action@v4
         with:
           token: ${{ secrets.CODECOV_TOKEN }}
-          # Comma-separated list of files to upload
           files: coverage.xml

Justfile

@@ -5,11 +5,14 @@ default_functest_suite := 'tests/functionals'
 install:
     uv sync --group dev
 
-test: lint unittest
+test: lint typecheck unittest
 
 lint:
     uv run ruff check .
 
+typecheck:
+    uv run mypy src/ tests/
+
 unittest test_suite=default_unittest_suite:
     uv run pytest -sxv {{test_suite}}
 

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,16 +33,39 @@ 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]
-dev = ["pytest >=8.3.3,<9", "pytest-asyncio >=0.24.0", "pytest-cov >=6.0.0,<7"]
+dev = [
+    "mypy>=1.13.0,<2",
+    "pytest >=8.3.3,<9",
+    "pytest-asyncio >=0.24.0",
+    "pytest-cov >=6.0.0,<7",
+]
+
+[tool.coverage.report]
+exclude_lines = [
+    "if TYPE_CHECKING:",
+    "except ImportError:",
+    "\\s+\\.\\.\\.$",
+    "# coverage: ignore",
+]
 
 [build-system]
 requires = ["pdm-backend"]

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)  # coverage: ignore
             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)