Resolve "Handle rate limit"

Author: LKaemmerling

hcloud/hcloud.py

@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 from __future__ import absolute_import
 
+import time
 import requests
 
 from hcloud.actions.client import ActionsClient
@@ -26,6 +27,7 @@ def __init__(self, code, message, details):
 
 class HcloudClient(object):
     version = VERSION
+    retry_wait_time = 0.5
 
     def __init__(self, token):
         self.token = token
@@ -68,7 +70,7 @@ def _raise_exception_from_json_content(self, json_content):
             details=json_content['error']['details']
         )
 
-    def request(self, method, url, **kwargs):
+    def request(self, method, url, tries=1, **kwargs):
         response = requests.request(
             method,
             self.api_endpoint + url,
@@ -85,7 +87,12 @@ def request(self, method, url, **kwargs):
 
         if not response.ok:
             if json_content:
-                self._raise_exception_from_json_content(json_content)
+                if json_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)
             else:
                 self._raise_exception_from_response(response)
 

tests/unit/test_hcloud.py

@@ -38,6 +38,19 @@ def fail_response(self, response):
         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"
@@ -114,3 +127,23 @@ def test_request_500_empty_content(self, mocked_requests, client, fail_response)
         assert error.code == 500
         assert error.message == "Internal Server Error"
         assert error.details["content"] == ""
+
+    def test_request_limit(self, mocked_requests, client, rate_limit_response):
+        client.retry_wait_time = 0
+        mocked_requests.request.return_value = rate_limit_response
+        with pytest.raises(HcloudAPIException) as exception_info:
+            client.request("POST", "http://url.com", params={"argument": "value"}, timeout=2)
+        error = exception_info.value
+        assert mocked_requests.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, mocked_requests, 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')
+        mocked_requests.request.side_effect = [rate_limit_response, response]
+
+        client.request("POST", "http://url.com", params={"argument": "value"}, timeout=2)
+        assert mocked_requests.request.call_count == 2