Partial typing annotation support + add mypy (#82)

* cb: rename bool callback to boolean

This avoids conflicting with bool type for further
type annotations

* code-style: add mypy

* code-style: mypy: fix no-untyped-call

* code-style: mypy: add simple types

* mypy: exclude setup.py

* mypy: fix 3.11/3.12 typing issue

* mypy: move tox dir to avoid type issues

* tests: increase min duration in reports

* mypy: add setuptools types

---------

Co-authored-by: Mathias Brulatout <[email protected]>

.pre-commit-config.yaml

@@ -24,3 +24,9 @@ repos:
         entry: pylint -rn -sn # Only display messages, don't display the score
         language: system
         types: [python]
+
+      - id: mypy
+        name: mypy
+        language: system
+        entry: mypy --non-interactive --install-types
+        types: [python]

conftest.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import collections
 import json
 import os
@@ -8,6 +10,8 @@
 import docker
 import pytest
 import requests
+from docker import DockerClient
+from docker.errors import APIError, NotFound
 from requests import RequestException
 
 CONSUL_VERSIONS = ["1.16.1", "1.17.3"]
@@ -19,7 +23,7 @@
 os.makedirs(LOGS_DIR, exist_ok=True)
 
 
-def get_free_ports(num, host=None):
+def get_free_ports(num: int, host=None) -> list[int]:
     if not host:
         host = "127.0.0.1"
     sockets = []
@@ -40,7 +44,7 @@ def _unset_consul_token():
         del os.environ["CONSUL_HTTP_TOKEN"]
 
 
-def start_consul_container(version, acl_master_token=None):
+def start_consul_container(version: str, acl_master_token: str | None = None):
     """
     Starts a Consul container. If acl_master_token is None, ACL will be disabled
     for this server, otherwise it will be enabled and the master token will be
@@ -87,22 +91,29 @@ def start_consul_container(version, acl_master_token=None):
             "acl": {"enabled": True, "tokens": {"initial_management": acl_master_token}},
         }
         merged_config = {**base_config, **acl_config}
-        docker_config["environment"]["CONSUL_LOCAL_CONFIG"] = json.dumps(merged_config)
-
-    def start_consul_container_with_retry(client, command, version, docker_config, max_retries=3, retry_delay=2):  # pylint: disable=inconsistent-return-statements
+        docker_config["environment"]["CONSUL_LOCAL_CONFIG"] = json.dumps(merged_config)  # type: ignore
+
+    def start_consul_container_with_retry(  # pylint: disable=inconsistent-return-statements
+        client: DockerClient,
+        command: str,
+        version: str,
+        docker_config: dict,
+        max_retries: int = 3,
+        retry_delay: int = 2,
+    ):
         """
         Start a Consul container with retries as a few initial attempts sometimes fail.
         """
         for attempt in range(max_retries):
             try:
                 container = client.containers.run(f"hashicorp/consul:{version}", command=command, **docker_config)
                 return container
-            except docker.errors.APIError:
+            except APIError:
                 # Cleanup that stray container as it might cause a naming conflict
                 try:
                     container = client.containers.get(docker_config["name"])
                     container.remove(force=True)
-                except docker.errors.NotFound:
+                except NotFound:
                     pass
                 if attempt == max_retries - 1:
                     raise
@@ -146,13 +157,13 @@ def start_consul_container_with_retry(client, command, version, docker_config, m
     raise Exception("Failed to verify Consul startup")  # pylint: disable=broad-exception-raised
 
 
-def get_consul_version(port):
+def get_consul_version(port: int) -> str:
     base_uri = f"http://127.0.0.1:{port}/v1/"
     response = requests.get(base_uri + "agent/self", timeout=10)
     return response.json()["Config"]["Version"].strip()
 
 
-def setup_and_teardown_consul(request, version, acl_master_token=None):
+def setup_and_teardown_consul(request, version, acl_master_token: str | None = None):
     # Start the container, yield, get container logs, store them in logs/<test_name>.log, stop the container
     container, port = start_consul_container(version=version, acl_master_token=acl_master_token)
     version = get_consul_version(port)

consul/aio.py

@@ -11,7 +11,7 @@
 class HTTPClient(base.HTTPClient):
     """Asyncio adapter for python consul using aiohttp library"""
 
-    def __init__(self, *args, loop=None, connections_limit=None, connections_timeout=None, **kwargs):
+    def __init__(self, *args, loop=None, connections_limit=None, connections_timeout=None, **kwargs) -> None:
         super().__init__(*args, **kwargs)
         self._loop = loop or asyncio.get_event_loop()
         connector_kwargs = {}
@@ -22,7 +22,7 @@ def __init__(self, *args, loop=None, connections_limit=None, connections_timeout
         if connections_timeout:
             timeout = aiohttp.ClientTimeout(total=connections_timeout)
             session_kwargs["timeout"] = timeout
-        self._session = aiohttp.ClientSession(connector=connector, **session_kwargs)
+        self._session = aiohttp.ClientSession(connector=connector, **session_kwargs)  # type: ignore
 
     async def _request(
         self, callback, method, uri, headers: Optional[Dict[str, str]], data=None, connections_timeout=None
@@ -31,7 +31,7 @@ async def _request(
         if connections_timeout:
             timeout = aiohttp.ClientTimeout(total=connections_timeout)
             session_kwargs["timeout"] = timeout
-        resp = await self._session.request(method, uri, headers=headers, data=data, **session_kwargs)
+        resp = await self._session.request(method, uri, headers=headers, data=data, **session_kwargs)  # type: ignore
         body = await resp.text(encoding="utf-8")
         if resp.status == 599:
             raise Timeout
@@ -43,7 +43,13 @@ def get(self, callback, path, params=None, headers: Optional[Dict[str, str]] = N
         return self._request(callback, "GET", uri, headers=headers, connections_timeout=connections_timeout)
 
     def put(
-        self, callback, path, params=None, data="", headers: Optional[Dict[str, str]] = None, connections_timeout=None
+        self,
+        callback,
+        path,
+        params=None,
+        data: str = "",
+        headers: Optional[Dict[str, str]] = None,
+        connections_timeout=None,
     ):
         uri = self.uri(path, params)
         return self._request(callback, "PUT", uri, headers=headers, data=data, connections_timeout=connections_timeout)
@@ -53,7 +59,13 @@ def delete(self, callback, path, params=None, headers: Optional[Dict[str, str]]
         return self._request(callback, "DELETE", uri, headers=headers, connections_timeout=connections_timeout)
 
     def post(
-        self, callback, path, params=None, data="", headers: Optional[Dict[str, str]] = None, connections_timeout=None
+        self,
+        callback,
+        path,
+        params=None,
+        data: str = "",
+        headers: Optional[Dict[str, str]] = None,
+        connections_timeout=None,
     ):
         uri = self.uri(path, params)
         return self._request(callback, "POST", uri, headers=headers, data=data, connections_timeout=connections_timeout)
@@ -63,13 +75,13 @@ def close(self):
 
 
 class Consul(base.Consul):
-    def __init__(self, *args, loop=None, connections_limit=None, connections_timeout=None, **kwargs):
+    def __init__(self, *args, loop=None, connections_limit=None, connections_timeout=None, **kwargs) -> None:
         self._loop = loop or asyncio.get_event_loop()
         self.connections_limit = connections_limit
         self.connections_timeout = connections_timeout
         super().__init__(*args, **kwargs)
 
-    def http_connect(self, host, port, scheme, verify=True, cert=None):
+    def http_connect(self, host: str, port: int, scheme, verify: bool = True, cert=None):
         return HTTPClient(
             host,
             port,

consul/api/acl/__init__.py

@@ -3,7 +3,7 @@
 
 
 class ACL:
-    def __init__(self, agent):
+    def __init__(self, agent) -> None:
         self.agent = agent
 
         self.token = self.tokens = Token(agent)

consul/api/acl/policy.py

@@ -1,46 +1,46 @@
+from __future__ import annotations
+
 import json
+from typing import Optional
 
 from consul.callback import CB
 
 
 class Policy:
-    def __init__(self, agent):
+    def __init__(self, agent) -> None:
         self.agent = agent
 
-    def list(self, token=None):
+    def list(self, token: str | None = None):
         """
         Lists all the active ACL policies. This is a privileged endpoint, and
         requires a management token. *token* will override this client's
         default token.
         Requires a token with acl:read capability. ACLPermissionDenied raised otherwise
         """
-        params = []
 
         headers = self.agent.prepare_headers(token)
-        return self.agent.http.get(CB.json(), "/v1/acl/policies", params=params, headers=headers)
+        return self.agent.http.get(CB.json(), "/v1/acl/policies", headers=headers)
 
-    def read(self, uuid, token=None):
+    def read(self, uuid, token: str | None = None):
         """
         Returns the policy information for *id*. Requires a token with acl:read capability.
-        :param accessor_id: Specifies the UUID of the policy you lookup.
+        :param uuid: Specifies the UUID of the policy you look up.
         :param token: token with acl:read capability
         :return: selected Polic information
         """
-        params = []
         headers = self.agent.prepare_headers(token)
-        return self.agent.http.get(CB.json(), f"/v1/acl/policy/{uuid}", params=params, headers=headers)
+        return self.agent.http.get(CB.json(), f"/v1/acl/policy/{uuid}", headers=headers)
 
-    def create(self, name, token=None, description=None, rules=None):
+    def create(self, name: str, token: str | None = None, description: Optional[str] = None, rules=None):
         """
         Create a policy
         This is a privileged endpoint, and requires a token with acl:write.
         :param name: Specifies a name for the ACL policy.
         :param token: token with acl:write capability
-        :param description: Free form human readable description of the policy.
+        :param description: Free form human-readable description of the policy.
         :param rules: Specifies rules for the ACL policy.
         :return: The cloned token information
         """
-        params = []
         json_data = {"name": name}
         if rules:
             json_data["rules"] = json.dumps(rules)
@@ -50,7 +50,6 @@ def create(self, name, token=None, description=None, rules=None):
         return self.agent.http.put(
             CB.json(),
             "/v1/acl/policy",
-            params=params,
             headers=headers,
             data=json.dumps(json_data),
         )

consul/api/acl/token.py

@@ -1,79 +1,83 @@
+from __future__ import annotations
+
 import json
+import typing
 
 from consul.callback import CB
 
 
 class Token:
-    def __init__(self, agent):
+    def __init__(self, agent) -> None:
         self.agent = agent
 
-    def list(self, token=None):
+    def list(self, token: str | None = None):
         """
         Lists all the active ACL tokens. This is a privileged endpoint, and
         requires a management token. *token* will override this client's
         default token.
         Requires a token with acl:read capability. ACLPermissionDenied raised otherwise
         """
-        params = []
         headers = self.agent.prepare_headers(token)
-        return self.agent.http.get(CB.json(), "/v1/acl/tokens", params=params, headers=headers)
+        return self.agent.http.get(CB.json(), "/v1/acl/tokens", headers=headers)
 
-    def read(self, accessor_id, token=None):
+    def read(self, accessor_id: str, token: str | None = None):
         """
         Returns the token information for *accessor_id*. Requires a token with acl:read capability.
         :param accessor_id: The accessor ID of the token to read
         :param token: token with acl:read capability
         :return: selected token information
         """
-        params = []
         headers = self.agent.prepare_headers(token)
-        return self.agent.http.get(CB.json(), f"/v1/acl/token/{accessor_id}", params=params, headers=headers)
+        return self.agent.http.get(CB.json(), f"/v1/acl/token/{accessor_id}", headers=headers)
 
-    def delete(self, accessor_id, token=None):
+    def delete(self, accessor_id: str, token: str | None = None):
         """
         Deletes the token with *accessor_id*. This is a privileged endpoint, and requires a token with acl:write.
         :param accessor_id: The accessor ID of the token to delete
         :param token: token with acl:write capability
         :return: True if the token was deleted
         """
-        params = []
         headers = self.agent.prepare_headers(token)
-        return self.agent.http.delete(CB.bool(), f"/v1/acl/token/{accessor_id}", params=params, headers=headers)
+        return self.agent.http.delete(CB.boolean(), f"/v1/acl/token/{accessor_id}", headers=headers)
 
-    def clone(self, accessor_id, token=None, description=""):
+    def clone(self, accessor_id: str, token: str | None = None, description: str = ""):
         """
         Clones the token identified by *accessor_id*. This is a privileged endpoint, and requires a token with acl:write.
         :param accessor_id: The accessor ID of the token to clone
         :param token: token with acl:write capability
         :param description: Optional new token description
         :return: The cloned token information
         """
-        params = []
 
         json_data = {"Description": description}
         headers = self.agent.prepare_headers(token)
         return self.agent.http.put(
             CB.json(),
             f"/v1/acl/token/{accessor_id}/clone",
-            params=params,
             headers=headers,
             data=json.dumps(json_data),
         )
 
-    def create(self, token=None, accessor_id=None, secret_id=None, policies_id=None, description=""):
+    def create(
+        self,
+        token: str | None = None,
+        accessor_id: str | None = None,
+        secret_id: str | None = None,
+        policies_id: typing.List[str] | None = None,
+        description: str = "",
+    ):
         """
         Create a token (optionally identified by *secret_id* and *accessor_id*).
         This is a privileged endpoint, and requires a token with acl:write.
         :param token: token with acl:write capability
         :param accessor_id: The accessor ID of the token to create
         :param secret_id: The secret ID of the token to create
         :param description: Optional new token description
-        :param policies: Optional list of policies id
+        :param policies_id: Optional list of policies id
         :return: The cloned token information
         """
-        params = []
 
-        json_data = {}
+        json_data: dict[str, typing.Any] = {}
         if accessor_id:
             json_data["AccessorID"] = accessor_id
         if secret_id:
@@ -87,12 +91,11 @@ def create(self, token=None, accessor_id=None, secret_id=None, policies_id=None,
         return self.agent.http.put(
             CB.json(),
             "/v1/acl/token",
-            params=params,
             headers=headers,
             data=json.dumps(json_data),
         )
 
-    def update(self, accessor_id, token=None, secret_id=None, description=""):
+    def update(self, accessor_id: str, token: str | None = None, secret_id: str | None = None, description: str = ""):
         """
         Update a token (optionally identified by *secret_id* and *accessor_id*).
         This is a privileged endpoint, and requires a token with acl:write.
@@ -102,7 +105,6 @@ def update(self, accessor_id, token=None, secret_id=None, description=""):
         :param description: Optional new token description
         :return: The updated token information
         """
-        params = []
 
         json_data = {"AccessorID": accessor_id}
         if secret_id:
@@ -113,7 +115,6 @@ def update(self, accessor_id, token=None, secret_id=None, description=""):
         return self.agent.http.put(
             CB.json(),
             f"/v1/acl/token/{accessor_id}",
-            params=params,
             headers=headers,
             data=json.dumps(json_data),
         )