Improve typing coverage on the api_client module (#1347)

Author: therve

.generator/src/generator/templates/api_client.j2

@@ -6,24 +6,25 @@ import mimetypes
 import warnings
 import multiprocessing
 from multiprocessing.pool import ThreadPool
+from datetime import date, datetime
 import io
 import os
 import re
 from typing import Any, Dict, Optional, List, Tuple, Union
+from typing_extensions import Self
 from urllib.parse import quote
 from urllib3.fields import RequestField  # type: ignore
 
 
 from {{ package }} import rest
+from {{ package }}.configuration import Configuration
 from {{ package }}.exceptions import ApiTypeError, ApiValueError
 from {{ package }}.model_utils import (
     ModelNormal,
     ModelSimple,
     ModelComposed,
     check_allowed_values,
     check_validations,
-    date,
-    datetime,
     deserialize_file,
     file_type,
     model_to_dict,
@@ -32,7 +33,7 @@ from {{ package }}.model_utils import (
 )
 
 
-class ApiClient(object):
+class ApiClient:
     """Generic API client for OpenAPI client library builds.
 
     OpenAPI generic API client. This client handles the client-
@@ -46,7 +47,7 @@ class ApiClient(object):
         the API.
     """
 
-    def __init__(self, configuration):
+    def __init__(self, configuration: Configuration):
         self.configuration = configuration
 
         self.rest_client = self._build_rest_client()
@@ -56,28 +57,28 @@ class ApiClient(object):
         # Set default User-Agent.
         self.user_agent = user_agent()
 
-    def __enter__(self):
+    def __enter__(self) -> Self:
         return self
 
-    def __exit__(self, exc_type, exc_value, traceback):
+    def __exit__(self, exc_type, exc_value, traceback) -> None:
         self.close()
 
-    def close(self):
+    def close(self) -> None:
         self.rest_client.pool_manager.clear()
 
     def _build_rest_client(self):
         return rest.RESTClientObject(self.configuration)
 
     @property
-    def user_agent(self):
+    def user_agent(self) -> str:
         """User agent for this API client"""
         return self.default_headers["User-Agent"]
 
     @user_agent.setter
-    def user_agent(self, value):
+    def user_agent(self, value: str) -> None:
         self.default_headers["User-Agent"] = value
 
-    def set_default_header(self, header_name, header_value):
+    def set_default_header(self, header_name: str, header_value: str) -> None:
         self.default_headers[header_name] = header_value
 
     def _call_api(
@@ -91,7 +92,7 @@ class ApiClient(object):
         response_type: Optional[Tuple[Any]] = None,
         return_http_data_only: Optional[bool] = None,
         preload_content: bool = True,
-        request_timeout: Optional[Union[int, float, Tuple]] = None,
+        request_timeout: Optional[Union[int, float, Tuple[Union[int, float], Union[int, float]]]] = None,
         check_type: Optional[bool] = None,
     ):
         # perform request and return response
@@ -133,19 +134,16 @@ class ApiClient(object):
             return return_data
         return (return_data, response.status, response.getheaders())
 
-    def parameters_to_multipart(self, params, collection_types):
-        """Get parameters as list of tuples, formatting as json if value is collection_types.
+    def parameters_to_multipart(self, params):
+        """Get parameters as list of tuples, formatting as json if value is dict.
 
         :param params: Parameters as list of two-tuples.
-        :param collection_types: Parameter collection types.
 
         :return: Parameters as list of tuple or urllib3.fields.RequestField
         """
         new_params = []
-        if collection_types is None:
-            collection_types = dict
         for k, v in params.items() if isinstance(params, dict) else params:
-            if isinstance(v, collection_types):  # v is instance of collection_type, formatting as application/json
+            if isinstance(v, dict):  # v is instance of collection_type, formatting as application/json
                 v = json.dumps(v, ensure_ascii=False).encode("utf-8")
                 field = RequestField(k, v)
                 field.make_multipart(content_type="application/json; charset=utf-8")
@@ -175,7 +173,7 @@ class ApiClient(object):
         elif isinstance(obj, (str, int, float, bool)) or obj is None:
             return obj
         elif isinstance(obj, (datetime, date)):
-            if obj.tzinfo is not None:
+            if getattr(obj, "tzinfo", None) is not None:
                 return obj.isoformat()
             return obj.strftime("%Y-%m-%dT%H:%M:%S") + obj.strftime(".%f")[:4] + "Z"
         elif isinstance(obj, ModelSimple):
@@ -186,7 +184,7 @@ class ApiClient(object):
             return {key: cls.sanitize_for_serialization(val) for key, val in obj.items()}
         raise ApiValueError("Unable to prepare type {} for serialization".format(obj.__class__.__name__))
 
-    def deserialize(self, response_data, response_type, check_type):
+    def deserialize(self, response_data: str, response_type: Any, check_type: Optional[bool]):
         """Deserializes response into an object.
 
         :param response_data: Response data to be deserialized.
@@ -233,7 +231,7 @@ class ApiClient(object):
         return_http_data_only: Optional[bool] = None,
         collection_formats: Optional[Dict[str, str]] = None,
         preload_content: bool = True,
-        request_timeout: Optional[Union[int, float, Tuple]] = None,
+        request_timeout: Optional[Union[int, float, Tuple[Union[int, float], Union[int, float]]]] = None,
         host: Optional[str] = None,
         check_type: Optional[bool] = None,
     ):
@@ -310,7 +308,7 @@ class ApiClient(object):
             post_params = self.parameters_to_tuples(post_params, collection_formats)
             post_params.extend(self.files_parameters(files))
             if header_params["Content-Type"].startswith("multipart"):
-                post_params = self.parameters_to_multipart(post_params, (dict))
+                post_params = self.parameters_to_multipart(post_params)
 
         # body
         if body:
@@ -404,15 +402,15 @@ class ApiClient(object):
 
         return params
 
-    def select_header_accept(self, accepts):
+    def select_header_accept(self, accepts: List[str]) -> str:
         """Returns `Accept` based on an array of accepts provided.
 
         :param accepts: List of headers.
         :return: Accept (e.g. application/json).
         """
         return ", ".join(accepts)
 
-    def select_header_content_type(self, content_types):
+    def select_header_content_type(self, content_types: List[str]) -> str:
         """Returns `Content-Type` based on an array of content_types provided.
 
         :param content_types: List of content-types.
@@ -432,15 +430,15 @@ class ThreadedApiClient(ApiClient):
 
     _pool = None
 
-    def __init__(self, configuration, pool_threads=1):
+    def __init__(self, configuration: Configuration, pool_threads: int = 1):
         self.pool_threads = pool_threads
         self.connection_pool_maxsize = multiprocessing.cpu_count() * 5
         super().__init__(configuration)
 
     def _build_rest_client(self):
         return rest.RESTClientObject(self.configuration, maxsize=self.connection_pool_maxsize)
 
-    def close(self):
+    def close(self) -> None:
         self.rest_client.pool_manager.clear()
         if self._pool:
             self._pool.close()
@@ -450,7 +448,7 @@ class ThreadedApiClient(ApiClient):
                 atexit.unregister(self.close)
 
     @property
-    def pool(self):
+    def pool(self) -> ThreadPool:
         """Create thread pool on first request
         avoids instantiating unused threadpool for blocking clients.
         """
@@ -487,15 +485,15 @@ class ThreadedApiClient(ApiClient):
                 preload_content,
                 request_timeout,
                 check_type,
-            )
+            ),
         )
 
 
 class AsyncApiClient(ApiClient):
     def _build_rest_client(self):
         return rest.AsyncRESTClientObject(self.configuration)
 
-    async def __aenter__(self):
+    async def __aenter__(self) -> Self:
         return self
 
     async def __aexit__(self, _exc_type, exc, _tb):
@@ -514,7 +512,7 @@ class AsyncApiClient(ApiClient):
         response_type: Optional[Tuple[Any]] = None,
         return_http_data_only: Optional[bool] = None,
         preload_content: bool = True,
-        request_timeout: Optional[Union[int, float, Tuple]] = None,
+        request_timeout: Optional[Union[int, float, Tuple[Union[int, float], Union[int, float]]]] = None,
         check_type: Optional[bool] = None,
     ):
 
@@ -553,8 +551,14 @@ class AsyncApiClient(ApiClient):
         return (return_data, response.status_code, response.headers)
 
 
-class Endpoint(object):
-    def __init__(self, settings=None, params_map=None, headers_map=None, api_client=None):
+class Endpoint:
+    def __init__(
+        self,
+        settings: Dict[str, Any],
+        params_map: Dict[str, Dict[str, Any]],
+        headers_map: Dict[str, List[str]],
+        api_client: ApiClient,
+    ):
         """Creates an endpoint.
 
         :param settings: See below key value pairs:
@@ -646,17 +650,16 @@ class Endpoint(object):
     def call_with_http_info(self, **kwargs):
 
         is_unstable = self.api_client.configuration.unstable_operations.get(
-            "{}.{}".format(self.settings["version"], self.settings["operation_id"]))
+            "{}.{}".format(self.settings["version"], self.settings["operation_id"])
+        )
         if is_unstable:
             warnings.warn("Using unstable operation '{0}'".format(self.settings["operation_id"]))
         elif is_unstable is False:
             raise ApiValueError("Unstable operation '{0}' is disabled".format(self.settings["operation_id"]))
 
         try:
-            index = (
-                self.api_client.configuration.server_operation_index.get(
-                    self.settings["operation_id"], self.api_client.configuration.server_index
-                )
+            index = self.api_client.configuration.server_operation_index.get(
+                self.settings["operation_id"], self.api_client.configuration.server_index
             )
             server_variables = self.api_client.configuration.server_operation_variables.get(
                 self.settings["operation_id"], self.api_client.configuration.server_variables
@@ -677,7 +680,11 @@ class Endpoint(object):
             # only throw this nullable ApiValueError if check_input_type
             # is False, if check_input_type==True we catch this case
             # in self._validate_inputs
-            if not self.params_map[key].get("nullable") and value is None and not self.api_client.configuration.check_input_type:
+            if (
+                not self.params_map[key].get("nullable")
+                and value is None
+                and not self.api_client.configuration.check_input_type
+            ):
                 raise ApiValueError(
                     "Value may not be None for non-nullable parameter `%s`"
                     " when calling `%s`" % (key, self.settings["operation_id"])
@@ -722,7 +729,7 @@ class Endpoint(object):
             collection_formats=params["collection_format"],
         )
 
-    def update_params_for_auth(self, headers, queries):
+    def update_params_for_auth(self, headers, queries) -> None:
         """Updates header and query params based on authentication setting.
 
         :param headers: Header parameters dict to be updated.
@@ -745,7 +752,7 @@ class Endpoint(object):
                     raise ApiValueError("Authentication token must be in `query` or `header`")
 
 
-def user_agent():
+def user_agent() -> str:
     """Generate default User-Agent header."""
     import platform
     from {{ package }}.version import __version__