Adding annotations to errors, rest_client and client

plugins/module_utils/client.py

@@ -4,14 +4,17 @@
 # GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
 
 from __future__ import absolute_import, division, print_function
+from __future__ import annotations
 
 __metaclass__ = type
 
 import json
+from typing import Any, Optional, Union
 
 from ansible.module_utils.urls import Request, basic_auth_header
 
 from .errors import AuthError, ScaleComputingError, UnexpectedAPIResponse
+from ..module_utils.typed_classes import TypedClusterInstance
 
 from ansible.module_utils.six.moves.urllib.error import HTTPError, URLError
 from ansible.module_utils.six.moves.urllib.parse import urlencode, quote
@@ -25,7 +28,9 @@ class Response:
     # Response(raw_resp) would be simpler.
     # How is this used in other projects? Jure?
     # Maybe we need/want both.
-    def __init__(self, status, data, headers=None):
+    def __init__(
+        self, status: int, data: Any, headers: Optional[dict[Any, Any]] = None
+    ):
         self.status = status
         self.data = data
         # [('h1', 'v1'), ('H2', 'V2')] -> {'h1': 'v1', 'h2': 'V2'}
@@ -36,7 +41,7 @@ def __init__(self, status, data, headers=None):
         self._json = None
 
     @property
-    def json(self):
+    def json(self) -> Any:
         if self._json is None:
             try:
                 self._json = json.loads(self.data)
@@ -66,11 +71,11 @@ def __init__(
         self.password = password
         self.timeout = timeout
 
-        self._auth_header = None
+        self._auth_header: Optional[dict[str, bytes]] = None
         self._client = Request()
 
     @classmethod
-    def get_client(cls, cluster_instance: dict):
+    def get_client(cls, cluster_instance: TypedClusterInstance) -> Client:
         return cls(
             cluster_instance["host"],
             cluster_instance["username"],
@@ -79,18 +84,25 @@ def get_client(cls, cluster_instance: dict):
         )
 
     @property
-    def auth_header(self):
+    def auth_header(self) -> dict[str, bytes]:
         if not self._auth_header:
             self._auth_header = self._login()
         return self._auth_header
 
-    def _login(self):
+    def _login(self) -> dict[str, bytes]:
         return self._login_username_password()
 
-    def _login_username_password(self):
+    def _login_username_password(self) -> dict[str, bytes]:
         return dict(Authorization=basic_auth_header(self.username, self.password))
 
-    def _request(self, method, path, data=None, headers=None, timeout=None):
+    def _request(
+        self,
+        method: str,
+        path: str,
+        data: Optional[Union[dict[Any, Any], bytes, str]] = None,
+        headers: Optional[dict[Any, Any]] = None,
+        timeout: Optional[float] = None,
+    ) -> Response:
         if (
             timeout is None
         ):  # If timeout from request is not specifically provided, take it from the Client.
@@ -134,14 +146,14 @@ def _request(self, method, path, data=None, headers=None, timeout=None):
 
     def request(
         self,
-        method,
-        path,
-        query=None,
-        data=None,
-        headers=None,
-        binary_data=None,
-        timeout=None,
-    ):
+        method: str,
+        path: str,
+        query: Optional[dict[Any, Any]] = None,
+        data: Optional[dict[Any, Any]] = None,
+        headers: Optional[dict[Any, Any]] = None,
+        binary_data: Optional[bytes] = None,
+        timeout: Optional[float] = None,
+    ) -> Response:
         # Make sure we only have one kind of payload
         if data is not None and binary_data is not None:
             raise AssertionError(
@@ -155,31 +167,64 @@ def request(
             url = "{0}?{1}".format(url, urlencode(query))
         headers = dict(headers or DEFAULT_HEADERS, **self.auth_header)
         if data is not None:
-            data = json.dumps(data, separators=(",", ":"))
             headers["Content-type"] = "application/json"
+            return self._request(
+                method,
+                url,
+                data=json.dumps(data, separators=(",", ":")),
+                headers=headers,
+                timeout=timeout,
+            )
         elif binary_data is not None:
-            data = binary_data
+            return self._request(
+                method, url, data=binary_data, headers=headers, timeout=timeout
+            )
         return self._request(method, url, data=data, headers=headers, timeout=timeout)
 
-    def get(self, path, query=None, timeout=None):
+    def get(
+        self,
+        path: str,
+        query: Optional[dict[Any, Any]] = None,
+        timeout: Optional[float] = None,
+    ) -> Request:
         resp = self.request("GET", path, query=query, timeout=timeout)
         if resp.status in (200, 404):
             return resp
         raise UnexpectedAPIResponse(response=resp)
 
-    def post(self, path, data, query=None, timeout=None):
+    def post(
+        self,
+        path: str,
+        data: Optional[dict[Any, Any]],
+        query: Optional[dict[Any, Any]] = None,
+        timeout: Optional[float] = None,
+    ) -> Request:
         resp = self.request("POST", path, data=data, query=query, timeout=timeout)
         if resp.status == 201 or resp.status == 200:
             return resp
         raise UnexpectedAPIResponse(response=resp)
 
-    def patch(self, path, data, query=None, timeout=None):
+    def patch(
+        self,
+        path: str,
+        data: dict[Any, Any],
+        query: Optional[dict[Any, Any]] = None,
+        timeout: Optional[float] = None,
+    ) -> Request:
         resp = self.request("PATCH", path, data=data, query=query, timeout=timeout)
         if resp.status == 200:
             return resp
         raise UnexpectedAPIResponse(response=resp)
 
-    def put(self, path, data, query=None, timeout=None, binary_data=None, headers=None):
+    def put(
+        self,
+        path: str,
+        data: dict[Any, Any],
+        query: Optional[dict[Any, Any]] = None,
+        timeout: Optional[float] = None,
+        binary_data: Optional[bytes] = None,
+        headers: Optional[dict[Any, Any]] = None,
+    ) -> Request:
         resp = self.request(
             "PUT",
             path,
@@ -193,7 +238,12 @@ def put(self, path, data, query=None, timeout=None, binary_data=None, headers=No
             return resp
         raise UnexpectedAPIResponse(response=resp)
 
-    def delete(self, path, query=None, timeout=None):
+    def delete(
+        self,
+        path: str,
+        query: Optional[dict[Any, Any]] = None,
+        timeout: Optional[float] = None,
+    ) -> Request:
         resp = self.request("DELETE", path, query=query, timeout=timeout)
         if resp.status == 204 or resp.status == 200:
             return resp

plugins/module_utils/errors.py

@@ -7,7 +7,8 @@
 
 __metaclass__ = type
 
-from typing import Any
+from typing import Union
+from ansible.module_utils.urls import Request
 
 
 class ScaleComputingError(Exception):
@@ -19,7 +20,7 @@ class AuthError(ScaleComputingError):
 
 
 class UnexpectedAPIResponse(ScaleComputingError):
-    def __init__(self, response):
+    def __init__(self, response: Request):
         self.message = "Unexpected response - {0} {1}".format(
             response.status, response.data
         )
@@ -28,73 +29,73 @@ def __init__(self, response):
 
 
 class InvalidUuidFormatError(ScaleComputingError):
-    def __init__(self, data):
+    def __init__(self, data: Union[str, Exception]):
         self.message = "Invalid UUID - {0}".format(data)
         super(InvalidUuidFormatError, self).__init__(self.message)
 
 
 # In-case function parameter is optional but required
 class MissingFunctionParameter(ScaleComputingError):
-    def __init__(self, data):
+    def __init__(self, data: Union[str, Exception]):
         self.message = "Missing parameter - {0}".format(data)
         super(MissingFunctionParameter, self).__init__(self.message)
 
 
 # In-case argument spec doesn't catch exception
 class MissingValueAnsible(ScaleComputingError):
-    def __init__(self, data):
+    def __init__(self, data: Union[str, Exception]):
         self.message = "Missing value - {0}".format(data)
         super(MissingValueAnsible, self).__init__(self.message)
 
 
 # In-case argument spec doesn't catch exception
 class MissingValueHypercore(ScaleComputingError):
-    def __init__(self, data: Any):
+    def __init__(self, data: Union[str, Exception]):
         self.message = "Missing values from hypercore API - {0}".format(data)
         super(MissingValueHypercore, self).__init__(self.message)
 
 
 class DeviceNotUnique(ScaleComputingError):
-    def __init__(self, data):
+    def __init__(self, data: Union[str, Exception]):
         self.message = "Device is not unique - {0} - already exists".format(data)
         super(DeviceNotUnique, self).__init__(self.message)
 
 
 class VMNotFound(ScaleComputingError):
-    def __init__(self, data):
+    def __init__(self, data: Union[str, Exception]):
         self.message = "Virtual machine - {0} - not found".format(data)
         super(VMNotFound, self).__init__(self.message)
 
 
 class ReplicationNotUnique(ScaleComputingError):
-    def __init__(self, data):
+    def __init__(self, data: Union[str, Exception]):
         self.message = (
             "There is already a replication on - {0} - virtual machine".format(data)
         )
         super(ReplicationNotUnique, self).__init__(self.message)
 
 
 class ClusterConnectionNotFound(ScaleComputingError):
-    def __init__(self, data):
+    def __init__(self, data: Union[str, Exception]):
         self.message = "No cluster connection found - {0}".format(data)
         super(ClusterConnectionNotFound, self).__init__(self.message)
 
 
 class SMBServerNotFound(ScaleComputingError):
-    def __init__(self, data):
+    def __init__(self, data: Union[str, Exception]):
         self.message = "SMB server is either not connected or not in the same network - {0}".format(
             data
         )
         super(SMBServerNotFound, self).__init__(self.message)
 
 
 class VMInvalidParams(ScaleComputingError):
-    def __init__(self):
+    def __init__(self) -> None:
         self.message = "Invalid set of parameters - strict affinity set to true and nodes not provided."
         super(VMInvalidParams, self).__init__(self.message)
 
 
 class SupportTunnelError(ScaleComputingError):
-    def __init__(self, data):
+    def __init__(self, data: Union[str, Exception]):
         self.message = "{0}".format(data)
         super(SupportTunnelError, self).__init__(self.message)

plugins/module_utils/hypercore_version.py

@@ -334,5 +334,5 @@ def __eq__(self, other: object) -> bool:
 
     @classmethod
     def get(cls, rest_client: RestClient, check_mode: bool = False) -> UpdateStatus:
-        update_status = rest_client.client.get("update/update_status.json").json  # type: ignore
+        update_status = rest_client.client.get("update/update_status.json").json
         return cls.from_hypercore(update_status)