refactor: improve hcloud.Client (#227)

* refactor: simplify _get_user_agent

* refactor: rename json_content variable

* refactor: add typings to hcloud.Client

* test: modernize test_request_ok

Author: jooola

hcloud/hcloud.py

@@ -1,4 +1,5 @@
 import time
+from typing import Optional, Union
 
 import requests
 
@@ -32,24 +33,19 @@ class Client:
 
     def __init__(
         self,
-        token,
-        api_endpoint="https://api.hetzner.cloud/v1",
-        application_name=None,
-        application_version=None,
-        poll_interval=1,
+        token: str,
+        api_endpoint: str = "https://api.hetzner.cloud/v1",
+        application_name: Optional[str] = None,
+        application_version: Optional[str] = None,
+        poll_interval: int = 1,
     ):
         """Create an new Client instance
 
-        :param token: str
-                Hetzner Cloud API token
-        :param api_endpoint: str
-                Hetzner Cloud API endpoint (default is https://api.hetzner.cloud/v1)
-        :param application_name: str
-                Your application name (default is None)
-        :param application_version: str
-                Your application _version (default is None)
-        :param poll_interval: int
-                Interval for polling information from Hetzner Cloud API in seconds (default is 1)
+        :param token: Hetzner Cloud API token
+        :param api_endpoint: Hetzner Cloud API endpoint
+        :param application_name: Your application name
+        :param application_version: Your application _version
+        :param poll_interval: Interval for polling information from Hetzner Cloud API in seconds
         """
         self.token = token
         self._api_endpoint = api_endpoint
@@ -148,85 +144,79 @@ def __init__(
         :type: :class:`PlacementGroupsClient <hcloud.placement_groups.client.PlacementGroupsClient>`
         """
 
-    def _get_user_agent(self):
+    def _get_user_agent(self) -> str:
         """Get the user agent of the hcloud-python instance with the user application name (if specified)
 
-        :return: str
-            The user agent of this hcloud-python instance
-        """
-        if self._application_name is not None and self._application_version is None:
-            return "{application_name} {prefix}/{version}".format(
-                application_name=self._application_name,
-                prefix=self.__user_agent_prefix,
-                version=self._version,
-            )
-        elif (
-            self._application_name is not None and self._application_version is not None
-        ):
-            return "{application_name}/{application_version} {prefix}/{version}".format(
-                application_name=self._application_name,
-                application_version=self._application_version,
-                prefix=self.__user_agent_prefix,
-                version=self._version,
-            )
-        else:
-            return "{prefix}/{version}".format(
-                prefix=self.__user_agent_prefix, version=self._version
-            )
-
-    def _get_headers(self):
+        :return: The user agent of this hcloud-python instance
+        """
+        user_agents = []
+        for name, version in [
+            (self._application_name, self._application_version),
+            (self.__user_agent_prefix, self._version),
+        ]:
+            if name is not None:
+                user_agents.append(name if version is None else f"{name}/{version}")
+
+        return " ".join(user_agents)
+
+    def _get_headers(self) -> dict:
         headers = {
             "User-Agent": self._get_user_agent(),
             "Authorization": f"Bearer {self.token}",
         }
         return headers
 
-    def _raise_exception_from_response(self, response):
+    def _raise_exception_from_response(self, response: requests.Response):
         raise APIException(
             code=response.status_code,
             message=response.reason,
             details={"content": response.content},
         )
 
-    def _raise_exception_from_json_content(self, json_content):
+    def _raise_exception_from_content(self, content: dict):
         raise APIException(
-            code=json_content["error"]["code"],
-            message=json_content["error"]["message"],
-            details=json_content["error"]["details"],
+            code=content["error"]["code"],
+            message=content["error"]["message"],
+            details=content["error"]["details"],
         )
 
-    def request(self, method, url, tries=1, **kwargs):
+    def request(
+        self,
+        method: str,
+        url: str,
+        tries: int = 1,
+        **kwargs,
+    ) -> Union[bytes, dict]:
         """Perform a request to the Hetzner Cloud API, wrapper around requests.request
 
-        :param method: str
-                HTTP Method to perform the Request
-        :param url: str
-                URL of the Endpoint
-        :param tries: int
-                Tries of the request (used internally, should not be set by the user)
+        :param method: HTTP Method to perform the Request
+        :param url: URL of the Endpoint
+        :param tries: Tries of the request (used internally, should not be set by the user)
         :return: Response
-        :rtype: requests.Response
         """
         response = self._requests_session.request(
-            method, self._api_endpoint + url, headers=self._get_headers(), **kwargs
+            method=method,
+            url=self._api_endpoint + url,
+            headers=self._get_headers(),
+            **kwargs,
         )
 
-        json_content = response.content
+        content = response.content
         try:
-            if len(json_content) > 0:
-                json_content = response.json()
+            if len(content) > 0:
+                content = response.json()
         except (TypeError, ValueError):
             self._raise_exception_from_response(response)
 
         if not response.ok:
-            if json_content:
-                if json_content["error"]["code"] == "rate_limit_exceeded" and tries < 5:
+            if content:
+                if content["error"]["code"] == "rate_limit_exceeded" and tries < 5:
                     time.sleep(tries * self._retry_wait_time)
                     tries = tries + 1
                     return self.request(method, url, tries, **kwargs)
                 else:
-                    self._raise_exception_from_json_content(json_content)
+                    self._raise_exception_from_content(content)
             else:
                 self._raise_exception_from_response(response)
 
-        return json_content
+        return content

tests/unit/test_hcloud.py

@@ -81,15 +81,16 @@ def test_request_ok(self, client, response):
         response = client.request(
             "POST", "/servers", params={"argument": "value"}, timeout=2
         )
-        client._requests_session.request.assert_called_once()
-        assert client._requests_session.request.call_args[0] == (
-            "POST",
-            "https://api.hetzner.cloud/v1/servers",
+        client._requests_session.request.assert_called_once_with(
+            method="POST",
+            url="https://api.hetzner.cloud/v1/servers",
+            headers={
+                "User-Agent": "hcloud-python/0.0.0",
+                "Authorization": "Bearer project_token",
+            },
+            params={"argument": "value"},
+            timeout=2,
         )
-        assert client._requests_session.request.call_args[1]["params"] == {
-            "argument": "value"
-        }
-        assert client._requests_session.request.call_args[1]["timeout"] == 2
         assert response == {"result": "data"}
 
     def test_request_fails(self, client, fail_response):