From 413472d7af1602b872a9b56324b9bffd0067eee6 Mon Sep 17 00:00:00 2001 From: Jonas L Date: Fri, 14 Jul 2023 14:41:46 +0200 Subject: [PATCH] feat: move hcloud.hcloud module to hcloud._client (#243) * feat: move hcloud.hcloud module to hcloud._client * feat: add deprecation for hcloud.hcloud module --- hcloud/__init__.py | 2 +- hcloud/_client.py | 222 +++++++++++++++++++++++++++++++++++++ hcloud/hcloud.py | 227 ++------------------------------------ tests/unit/conftest.py | 2 +- tests/unit/test_client.py | 182 ++++++++++++++++++++++++++++++ tests/unit/test_hcloud.py | 182 +----------------------------- 6 files changed, 416 insertions(+), 401 deletions(-) create mode 100644 hcloud/_client.py create mode 100644 tests/unit/test_client.py diff --git a/hcloud/__init__.py b/hcloud/__init__.py index 592ff64..5beda91 100644 --- a/hcloud/__init__.py +++ b/hcloud/__init__.py @@ -1,2 +1,2 @@ +from ._client import Client # noqa from ._exceptions import APIException, HCloudException # noqa -from .hcloud import Client # noqa diff --git a/hcloud/_client.py b/hcloud/_client.py new file mode 100644 index 0000000..e62624f --- /dev/null +++ b/hcloud/_client.py @@ -0,0 +1,222 @@ +import time +from typing import Optional, Union + +import requests + +from .__version__ import VERSION +from ._exceptions import APIException +from .actions.client import ActionsClient +from .certificates.client import CertificatesClient +from .datacenters.client import DatacentersClient +from .firewalls.client import FirewallsClient +from .floating_ips.client import FloatingIPsClient +from .images.client import ImagesClient +from .isos.client import IsosClient +from .load_balancer_types.client import LoadBalancerTypesClient +from .load_balancers.client import LoadBalancersClient +from .locations.client import LocationsClient +from .networks.client import NetworksClient +from .placement_groups.client import PlacementGroupsClient +from .primary_ips.client import PrimaryIPsClient +from .server_types.client import ServerTypesClient +from .servers.client import ServersClient +from .ssh_keys.client import SSHKeysClient +from .volumes.client import VolumesClient + + +class Client: + """Base Client for accessing the Hetzner Cloud API""" + + _version = VERSION + _retry_wait_time = 0.5 + __user_agent_prefix = "hcloud-python" + + def __init__( + self, + 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: 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 + self._application_name = application_name + self._application_version = application_version + self._requests_session = requests.Session() + self.poll_interval = poll_interval + + self.datacenters = DatacentersClient(self) + """DatacentersClient Instance + + :type: :class:`DatacentersClient ` + """ + self.locations = LocationsClient(self) + """LocationsClient Instance + + :type: :class:`LocationsClient ` + """ + self.servers = ServersClient(self) + """ServersClient Instance + + :type: :class:`ServersClient ` + """ + self.server_types = ServerTypesClient(self) + """ServerTypesClient Instance + + :type: :class:`ServerTypesClient ` + """ + self.volumes = VolumesClient(self) + """VolumesClient Instance + + :type: :class:`VolumesClient ` + """ + self.actions = ActionsClient(self) + """ActionsClient Instance + + :type: :class:`ActionsClient ` + """ + self.images = ImagesClient(self) + """ImagesClient Instance + + :type: :class:`ImagesClient ` + """ + self.isos = IsosClient(self) + """ImagesClient Instance + + :type: :class:`IsosClient ` + """ + self.ssh_keys = SSHKeysClient(self) + """SSHKeysClient Instance + + :type: :class:`SSHKeysClient ` + """ + self.floating_ips = FloatingIPsClient(self) + """FloatingIPsClient Instance + + :type: :class:`FloatingIPsClient ` + """ + self.primary_ips = PrimaryIPsClient(self) + """PrimaryIPsClient Instance + + :type: :class:`PrimaryIPsClient ` + """ + self.networks = NetworksClient(self) + """NetworksClient Instance + + :type: :class:`NetworksClient ` + """ + self.certificates = CertificatesClient(self) + """CertificatesClient Instance + + :type: :class:`CertificatesClient ` + """ + + self.load_balancers = LoadBalancersClient(self) + """LoadBalancersClient Instance + + :type: :class:`LoadBalancersClient ` + """ + + self.load_balancer_types = LoadBalancerTypesClient(self) + """LoadBalancerTypesClient Instance + + :type: :class:`LoadBalancerTypesClient ` + """ + + self.firewalls = FirewallsClient(self) + """FirewallsClient Instance + + :type: :class:`FirewallsClient ` + """ + + self.placement_groups = PlacementGroupsClient(self) + """PlacementGroupsClient Instance + + :type: :class:`PlacementGroupsClient ` + """ + + def _get_user_agent(self) -> str: + """Get the user agent of the hcloud-python instance with the user application name (if specified) + + :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: requests.Response): + raise APIException( + code=response.status_code, + message=response.reason, + details={"content": response.content}, + ) + + def _raise_exception_from_content(self, content: dict): + raise APIException( + code=content["error"]["code"], + message=content["error"]["message"], + details=content["error"]["details"], + ) + + 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: 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 + """ + response = self._requests_session.request( + method=method, + url=self._api_endpoint + url, + headers=self._get_headers(), + **kwargs, + ) + + content = response.content + try: + if len(content) > 0: + content = response.json() + except (TypeError, ValueError): + self._raise_exception_from_response(response) + + if not response.ok: + 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_content(content) + else: + self._raise_exception_from_response(response) + + return content diff --git a/hcloud/hcloud.py b/hcloud/hcloud.py index e62624f..af4e5c2 100644 --- a/hcloud/hcloud.py +++ b/hcloud/hcloud.py @@ -1,222 +1,9 @@ -import time -from typing import Optional, Union +import warnings -import requests +warnings.warn( + "The 'hcloud.hcloud' module is deprecated, please import from the 'hcloud' module instead (e.g. 'from hcloud import Client').", + DeprecationWarning, + stacklevel=2, +) -from .__version__ import VERSION -from ._exceptions import APIException -from .actions.client import ActionsClient -from .certificates.client import CertificatesClient -from .datacenters.client import DatacentersClient -from .firewalls.client import FirewallsClient -from .floating_ips.client import FloatingIPsClient -from .images.client import ImagesClient -from .isos.client import IsosClient -from .load_balancer_types.client import LoadBalancerTypesClient -from .load_balancers.client import LoadBalancersClient -from .locations.client import LocationsClient -from .networks.client import NetworksClient -from .placement_groups.client import PlacementGroupsClient -from .primary_ips.client import PrimaryIPsClient -from .server_types.client import ServerTypesClient -from .servers.client import ServersClient -from .ssh_keys.client import SSHKeysClient -from .volumes.client import VolumesClient - - -class Client: - """Base Client for accessing the Hetzner Cloud API""" - - _version = VERSION - _retry_wait_time = 0.5 - __user_agent_prefix = "hcloud-python" - - def __init__( - self, - 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: 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 - self._application_name = application_name - self._application_version = application_version - self._requests_session = requests.Session() - self.poll_interval = poll_interval - - self.datacenters = DatacentersClient(self) - """DatacentersClient Instance - - :type: :class:`DatacentersClient ` - """ - self.locations = LocationsClient(self) - """LocationsClient Instance - - :type: :class:`LocationsClient ` - """ - self.servers = ServersClient(self) - """ServersClient Instance - - :type: :class:`ServersClient ` - """ - self.server_types = ServerTypesClient(self) - """ServerTypesClient Instance - - :type: :class:`ServerTypesClient ` - """ - self.volumes = VolumesClient(self) - """VolumesClient Instance - - :type: :class:`VolumesClient ` - """ - self.actions = ActionsClient(self) - """ActionsClient Instance - - :type: :class:`ActionsClient ` - """ - self.images = ImagesClient(self) - """ImagesClient Instance - - :type: :class:`ImagesClient ` - """ - self.isos = IsosClient(self) - """ImagesClient Instance - - :type: :class:`IsosClient ` - """ - self.ssh_keys = SSHKeysClient(self) - """SSHKeysClient Instance - - :type: :class:`SSHKeysClient ` - """ - self.floating_ips = FloatingIPsClient(self) - """FloatingIPsClient Instance - - :type: :class:`FloatingIPsClient ` - """ - self.primary_ips = PrimaryIPsClient(self) - """PrimaryIPsClient Instance - - :type: :class:`PrimaryIPsClient ` - """ - self.networks = NetworksClient(self) - """NetworksClient Instance - - :type: :class:`NetworksClient ` - """ - self.certificates = CertificatesClient(self) - """CertificatesClient Instance - - :type: :class:`CertificatesClient ` - """ - - self.load_balancers = LoadBalancersClient(self) - """LoadBalancersClient Instance - - :type: :class:`LoadBalancersClient ` - """ - - self.load_balancer_types = LoadBalancerTypesClient(self) - """LoadBalancerTypesClient Instance - - :type: :class:`LoadBalancerTypesClient ` - """ - - self.firewalls = FirewallsClient(self) - """FirewallsClient Instance - - :type: :class:`FirewallsClient ` - """ - - self.placement_groups = PlacementGroupsClient(self) - """PlacementGroupsClient Instance - - :type: :class:`PlacementGroupsClient ` - """ - - def _get_user_agent(self) -> str: - """Get the user agent of the hcloud-python instance with the user application name (if specified) - - :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: requests.Response): - raise APIException( - code=response.status_code, - message=response.reason, - details={"content": response.content}, - ) - - def _raise_exception_from_content(self, content: dict): - raise APIException( - code=content["error"]["code"], - message=content["error"]["message"], - details=content["error"]["details"], - ) - - 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: 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 - """ - response = self._requests_session.request( - method=method, - url=self._api_endpoint + url, - headers=self._get_headers(), - **kwargs, - ) - - content = response.content - try: - if len(content) > 0: - content = response.json() - except (TypeError, ValueError): - self._raise_exception_from_response(response) - - if not response.ok: - 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_content(content) - else: - self._raise_exception_from_response(response) - - return content +from ._client import * # noqa diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index d15d36b..0ab7490 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -7,7 +7,7 @@ @pytest.fixture(autouse=True, scope="function") def mocked_requests(): - patcher = mock.patch("hcloud.hcloud.requests") + patcher = mock.patch("hcloud._client.requests") mocked_requests = patcher.start() yield mocked_requests patcher.stop() diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py new file mode 100644 index 0000000..5f5bf81 --- /dev/null +++ b/tests/unit/test_client.py @@ -0,0 +1,182 @@ +import json +from unittest.mock import MagicMock + +import pytest +import requests + +from hcloud import APIException, Client + + +class TestHetznerClient: + @pytest.fixture() + def client(self): + Client._version = "0.0.0" + client = Client(token="project_token") + + client._requests_session = MagicMock() + return client + + @pytest.fixture() + def response(self): + response = requests.Response() + response.status_code = 200 + response._content = json.dumps({"result": "data"}).encode("utf-8") + return response + + @pytest.fixture() + def fail_response(self, response): + response.status_code = 422 + error = { + "code": "invalid_input", + "message": "invalid input in field 'broken_field': is too long", + "details": { + "fields": [{"name": "broken_field", "messages": ["is too long"]}] + }, + } + response._content = json.dumps({"error": error}).encode("utf-8") + return response + + @pytest.fixture() + def rate_limit_response(self, response): + response.status_code = 422 + error = { + "code": "rate_limit_exceeded", + "message": "limit of 10 requests per hour reached", + "details": {}, + } + response._content = json.dumps({"error": error}).encode("utf-8") + return response + + def test__get_user_agent(self, client): + user_agent = client._get_user_agent() + assert user_agent == "hcloud-python/0.0.0" + + def test__get_user_agent_with_application_name(self, client): + client = Client(token="project_token", application_name="my-app") + user_agent = client._get_user_agent() + assert user_agent == "my-app hcloud-python/0.0.0" + + def test__get_user_agent_with_application_name_and_version(self, client): + client = Client( + token="project_token", + application_name="my-app", + application_version="1.0.0", + ) + user_agent = client._get_user_agent() + assert user_agent == "my-app/1.0.0 hcloud-python/0.0.0" + + def test__get_headers(self, client): + headers = client._get_headers() + assert headers == { + "User-Agent": "hcloud-python/0.0.0", + "Authorization": "Bearer project_token", + } + + def test_request_library_mocked(self, client): + response = client.request("POST", "url", params={"1": 2}) + assert response.__class__.__name__ == "MagicMock" + + def test_request_ok(self, client, response): + client._requests_session.request.return_value = response + response = client.request( + "POST", "/servers", params={"argument": "value"}, timeout=2 + ) + 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 response == {"result": "data"} + + def test_request_fails(self, client, fail_response): + client._requests_session.request.return_value = fail_response + with pytest.raises(APIException) as exception_info: + client.request( + "POST", "http://url.com", params={"argument": "value"}, timeout=2 + ) + error = exception_info.value + assert error.code == "invalid_input" + assert error.message == "invalid input in field 'broken_field': is too long" + assert error.details["fields"][0]["name"] == "broken_field" + + def test_request_500(self, client, fail_response): + fail_response.status_code = 500 + fail_response.reason = "Internal Server Error" + fail_response._content = "Internal Server Error" + client._requests_session.request.return_value = fail_response + with pytest.raises(APIException) as exception_info: + client.request( + "POST", "http://url.com", params={"argument": "value"}, timeout=2 + ) + error = exception_info.value + assert error.code == 500 + assert error.message == "Internal Server Error" + assert error.details["content"] == "Internal Server Error" + + def test_request_broken_json_200(self, client, response): + content = b"{'key': 'value'" + response.reason = "OK" + response._content = content + client._requests_session.request.return_value = response + with pytest.raises(APIException) as exception_info: + client.request( + "POST", "http://url.com", params={"argument": "value"}, timeout=2 + ) + error = exception_info.value + assert error.code == 200 + assert error.message == "OK" + assert error.details["content"] == content + + def test_request_empty_content_200(self, client, response): + content = "" + response.reason = "OK" + response._content = content + client._requests_session.request.return_value = response + response = client.request( + "POST", "http://url.com", params={"argument": "value"}, timeout=2 + ) + assert response == "" + + def test_request_500_empty_content(self, client, fail_response): + fail_response.status_code = 500 + fail_response.reason = "Internal Server Error" + fail_response._content = "" + client._requests_session.request.return_value = fail_response + with pytest.raises(APIException) as exception_info: + client.request( + "POST", "http://url.com", params={"argument": "value"}, timeout=2 + ) + error = exception_info.value + assert error.code == 500 + assert error.message == "Internal Server Error" + assert error.details["content"] == "" + assert str(error) == "Internal Server Error" + + def test_request_limit(self, client, rate_limit_response): + client._retry_wait_time = 0 + client._requests_session.request.return_value = rate_limit_response + with pytest.raises(APIException) as exception_info: + client.request( + "POST", "http://url.com", params={"argument": "value"}, timeout=2 + ) + error = exception_info.value + assert client._requests_session.request.call_count == 5 + assert error.code == "rate_limit_exceeded" + assert error.message == "limit of 10 requests per hour reached" + + def test_request_limit_then_success(self, client, rate_limit_response): + client._retry_wait_time = 0 + response = requests.Response() + response.status_code = 200 + response._content = json.dumps({"result": "data"}).encode("utf-8") + client._requests_session.request.side_effect = [rate_limit_response, response] + + client.request( + "POST", "http://url.com", params={"argument": "value"}, timeout=2 + ) + assert client._requests_session.request.call_count == 2 diff --git a/tests/unit/test_hcloud.py b/tests/unit/test_hcloud.py index 5f5bf81..87ab50a 100644 --- a/tests/unit/test_hcloud.py +++ b/tests/unit/test_hcloud.py @@ -1,182 +1,6 @@ -import json -from unittest.mock import MagicMock - import pytest -import requests - -from hcloud import APIException, Client - - -class TestHetznerClient: - @pytest.fixture() - def client(self): - Client._version = "0.0.0" - client = Client(token="project_token") - - client._requests_session = MagicMock() - return client - - @pytest.fixture() - def response(self): - response = requests.Response() - response.status_code = 200 - response._content = json.dumps({"result": "data"}).encode("utf-8") - return response - - @pytest.fixture() - def fail_response(self, response): - response.status_code = 422 - error = { - "code": "invalid_input", - "message": "invalid input in field 'broken_field': is too long", - "details": { - "fields": [{"name": "broken_field", "messages": ["is too long"]}] - }, - } - response._content = json.dumps({"error": error}).encode("utf-8") - return response - - @pytest.fixture() - def rate_limit_response(self, response): - response.status_code = 422 - error = { - "code": "rate_limit_exceeded", - "message": "limit of 10 requests per hour reached", - "details": {}, - } - response._content = json.dumps({"error": error}).encode("utf-8") - return response - - def test__get_user_agent(self, client): - user_agent = client._get_user_agent() - assert user_agent == "hcloud-python/0.0.0" - - def test__get_user_agent_with_application_name(self, client): - client = Client(token="project_token", application_name="my-app") - user_agent = client._get_user_agent() - assert user_agent == "my-app hcloud-python/0.0.0" - - def test__get_user_agent_with_application_name_and_version(self, client): - client = Client( - token="project_token", - application_name="my-app", - application_version="1.0.0", - ) - user_agent = client._get_user_agent() - assert user_agent == "my-app/1.0.0 hcloud-python/0.0.0" - - def test__get_headers(self, client): - headers = client._get_headers() - assert headers == { - "User-Agent": "hcloud-python/0.0.0", - "Authorization": "Bearer project_token", - } - - def test_request_library_mocked(self, client): - response = client.request("POST", "url", params={"1": 2}) - assert response.__class__.__name__ == "MagicMock" - - def test_request_ok(self, client, response): - client._requests_session.request.return_value = response - response = client.request( - "POST", "/servers", params={"argument": "value"}, timeout=2 - ) - 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 response == {"result": "data"} - - def test_request_fails(self, client, fail_response): - client._requests_session.request.return_value = fail_response - with pytest.raises(APIException) as exception_info: - client.request( - "POST", "http://url.com", params={"argument": "value"}, timeout=2 - ) - error = exception_info.value - assert error.code == "invalid_input" - assert error.message == "invalid input in field 'broken_field': is too long" - assert error.details["fields"][0]["name"] == "broken_field" - - def test_request_500(self, client, fail_response): - fail_response.status_code = 500 - fail_response.reason = "Internal Server Error" - fail_response._content = "Internal Server Error" - client._requests_session.request.return_value = fail_response - with pytest.raises(APIException) as exception_info: - client.request( - "POST", "http://url.com", params={"argument": "value"}, timeout=2 - ) - error = exception_info.value - assert error.code == 500 - assert error.message == "Internal Server Error" - assert error.details["content"] == "Internal Server Error" - - def test_request_broken_json_200(self, client, response): - content = b"{'key': 'value'" - response.reason = "OK" - response._content = content - client._requests_session.request.return_value = response - with pytest.raises(APIException) as exception_info: - client.request( - "POST", "http://url.com", params={"argument": "value"}, timeout=2 - ) - error = exception_info.value - assert error.code == 200 - assert error.message == "OK" - assert error.details["content"] == content - - def test_request_empty_content_200(self, client, response): - content = "" - response.reason = "OK" - response._content = content - client._requests_session.request.return_value = response - response = client.request( - "POST", "http://url.com", params={"argument": "value"}, timeout=2 - ) - assert response == "" - - def test_request_500_empty_content(self, client, fail_response): - fail_response.status_code = 500 - fail_response.reason = "Internal Server Error" - fail_response._content = "" - client._requests_session.request.return_value = fail_response - with pytest.raises(APIException) as exception_info: - client.request( - "POST", "http://url.com", params={"argument": "value"}, timeout=2 - ) - error = exception_info.value - assert error.code == 500 - assert error.message == "Internal Server Error" - assert error.details["content"] == "" - assert str(error) == "Internal Server Error" - - def test_request_limit(self, client, rate_limit_response): - client._retry_wait_time = 0 - client._requests_session.request.return_value = rate_limit_response - with pytest.raises(APIException) as exception_info: - client.request( - "POST", "http://url.com", params={"argument": "value"}, timeout=2 - ) - error = exception_info.value - assert client._requests_session.request.call_count == 5 - assert error.code == "rate_limit_exceeded" - assert error.message == "limit of 10 requests per hour reached" - def test_request_limit_then_success(self, client, rate_limit_response): - client._retry_wait_time = 0 - response = requests.Response() - response.status_code = 200 - response._content = json.dumps({"result": "data"}).encode("utf-8") - client._requests_session.request.side_effect = [rate_limit_response, response] - client.request( - "POST", "http://url.com", params={"argument": "value"}, timeout=2 - ) - assert client._requests_session.request.call_count == 2 +def test_deprecated_hcloud_hcloud_module(): + with pytest.deprecated_call(): + from hcloud.hcloud import Client # noqa