Implement pagination

Author: LD250

hcloud/actions/client.py

@@ -9,19 +9,34 @@ class BoundAction(BoundModelBase):
 
 
 class ActionsClient(ClientEntityBase):
+    results_list_attribute_name = 'actions'
 
     def get_by_id(self, id):
-        # type: (int) -> actions.client.BoundAction
+        # type: (int) -> BoundAction
         response = self._client.request(url="/actions/{action_id}".format(action_id=id), method="GET")
         return BoundAction(self, response['action'])
 
-    def get_all(self, status=None, sort=None):
-        # type: # type: (Optional[List[str], Optional[List[str]]) -> List[BoundAction]
+    def get_list(self,
+                 status=None,    # type: Optional[List[str]]
+                 sort=None,      # type: Optional[List[str]]
+                 page=None,      # type: Optional[int]
+                 per_page=None,  # type: Optional[int]
+                 ):
+        # type: (...) -> PageResults[List[BoundAction]]
         params = {}
         if status is not None:
-            params.update({"status": status})
+            params["status"] = status
         if sort is not None:
-            params.update({"sort": sort})
+            params["sort"] = sort
+        if page is not None:
+            params["page"] = page
+        if per_page is not None:
+            params["per_page"] = per_page
 
         response = self._client.request(url="/actions", method="GET", params=params)
-        return [BoundAction(self, action_data) for action_data in response['actions']]
+        actions = [BoundAction(self, action_data) for action_data in response['actions']]
+        return self.add_meta_to_result(actions, response)
+
+    def get_all(self, status=None, sort=None):
+        # type: (Optional[List[str]], Optional[List[str]]) -> List[BoundAction]
+        return super(ActionsClient, self).get_all(status=status, sort=sort)

hcloud/core/client.py

@@ -1,9 +1,64 @@
 # -*- coding: utf-8 -*-
+from hcloud.core.domain import add_meta_to_result
+
+
 class ClientEntityBase(object):
+    max_per_page = 50
+    results_list_attribute_name = None
 
     def __init__(self, client):
         self._client = client
 
+    def is_list_attribute_implemented(self):
+        if self.results_list_attribute_name is None:
+            raise NotImplementedError(
+                "in order to get results list, 'results_list_attribute_name' attribute of {} has to be specified". format(self.__class__.__name__)
+            )
+
+    def add_meta_to_result(self,
+                           results,  # type: List[BoundModelBase]
+                           response  # type: json
+                           ):
+        # type: (...) -> PageResult
+        self.is_list_attribute_implemented()
+        return add_meta_to_result(results, response, self.results_list_attribute_name)
+
+    def _get_all(self,
+                 list_function,                # type: function
+                 results_list_attribute_name,  # type: str
+                 *args,
+                 **kwargs
+                 ):
+        # type (...) -> List[BoundModelBase]
+        page = 1
+
+        results = []
+
+        while page:
+            page_result = list_function(page=page, per_page=self.max_per_page, *args, **kwargs)
+            result = getattr(page_result, results_list_attribute_name)
+            if result:
+                results.extend(result)
+            meta = page_result.meta
+            if meta and meta.pagination and meta.pagination.next_page and meta.pagination.next_page:
+                page = meta.pagination.next_page
+            else:
+                page = None
+
+        return results
+
+    def get_all(self, *args, **kwargs):
+        # type: (...) -> List[BoundModelBase]
+        self.is_list_attribute_implemented()
+        return self._get_all(self.get_list, self.results_list_attribute_name, *args, **kwargs)
+
+    def get_actions(self, *args, **kwargs):
+        # type: (...) -> List[BoundModelBase]
+        if not hasattr(self, 'get_actions_list'):
+            raise ValueError('this endpoint does not support get_actions method')
+
+        return self._get_all(self.get_actions_list, 'actions', *args, **kwargs)
+
 
 class BoundModelBase(object):
     model = None

hcloud/core/domain.py

@@ -1,7 +1,14 @@
 # -*- coding: utf-8 -*-
+from collections import namedtuple
+
+
 class BaseDomain(object):
     __slots__ = ()
 
+
+class DomainIdentityMixin(object):
+    __slots__ = ()
+
     @property
     def id_or_name(self):
         if self.id is not None:
@@ -10,3 +17,56 @@ def id_or_name(self):
             return self.name
         else:
             raise ValueError("id or name must be set")
+
+
+class Pagination(BaseDomain):
+    __slots__ = (
+        "page",
+        "per_page",
+        "previous_page",
+        "next_page",
+        "last_page",
+        "total_entries",
+    )
+
+    def __init__(self, page, per_page, previous_page=None, next_page=None, last_page=None, total_entries=None):
+        self.page = page
+        self.per_page = per_page
+        self.previous_page = previous_page
+        self.next_page = next_page
+        self.last_page = last_page
+        self.total_entries = total_entries
+
+
+class Meta(BaseDomain):
+
+    __slots__ = (
+        "pagination",
+    )
+
+    def __init__(
+        self,
+        pagination=None,
+    ):
+        self.pagination = pagination
+
+    @classmethod
+    def parse_meta(cls, json_content):
+        meta = None
+        if json_content and "meta" in json_content:
+            meta = cls()
+            pagination_json = json_content['meta'].get("pagination")
+            if pagination_json:
+                pagination = Pagination(**pagination_json)
+                meta.pagination = pagination
+        return meta
+
+
+def add_meta_to_result(result, json_content, attr_name):
+    # type: (List[BoundModelBase], json, string) -> PageResult
+    class_name = 'PageResults{0}'.format(attr_name.capitalize())
+    PageResults = namedtuple(class_name, [attr_name, 'meta'])
+    return PageResults(**{
+        attr_name: result,
+        'meta': Meta.parse_meta(json_content)
+    })

hcloud/datacenters/client.py

@@ -16,7 +16,6 @@ def __init__(self, client, data):
 
         server_types = data.get("server_types")
         if server_types is not None:
-            DatacenterServerTypes
             available = [BoundServerType(client._client.server_types, {"id": server_type}, complete=False) for server_type in server_types['available']]
             supported = [BoundServerType(client._client.server_types, {"id": server_type}, complete=False) for server_type in server_types['supported']]
             available_for_migration = [BoundServerType(client._client.server_types, {"id": server_type}, complete=False) for server_type in server_types['available_for_migration']]
@@ -26,17 +25,35 @@ def __init__(self, client, data):
 
 
 class DatacentersClient(ClientEntityBase):
+    results_list_attribute_name = 'datacenters'
 
     def get_by_id(self, id):
-        # type: (int) -> datacenters.client.BoundDatacenter
+        # type: (int) -> BoundDatacenter
         response = self._client.request(url="/datacenters/{datacenter_id}".format(datacenter_id=id), method="GET")
         return BoundDatacenter(self, response['datacenter'])
 
-    def get_all(self, name=None):
-        # type: (Optional[str]) -> List[BoundAction]
+    def get_list(self,
+                 name=None,     # type: Optional[str]
+                 page=None,     # type: Optional[int]
+                 per_page=None  # type: Optional[int]
+                 ):
+        # type: (...) -> PageResults[List[BoundDatacenter], Meta]
         params = {}
         if name is not None:
-            params.update({"name": name})
+            params["name"] = name
+
+        if page is not None:
+            params['page'] = page
+
+        if per_page is not None:
+            params['per_page'] = per_page
 
         response = self._client.request(url="/datacenters", method="GET", params=params)
-        return [BoundDatacenter(self, datacenter_data) for datacenter_data in response['datacenters']]
+
+        datacenters = [BoundDatacenter(self, datacenter_data) for datacenter_data in response['datacenters']]
+
+        return self.add_meta_to_result(datacenters, response)
+
+    def get_all(self, name=None):
+        # type: (Optional[str]) -> List[BoundDatacenter]
+        return super(DatacentersClient, self).get_all(name=name)

hcloud/datacenters/domain.py

@@ -1,8 +1,8 @@
 # -*- coding: utf-8 -*-
-from hcloud.core.domain import BaseDomain
+from hcloud.core.domain import BaseDomain, DomainIdentityMixin
 
 
-class Datacenter(BaseDomain):
+class Datacenter(BaseDomain, DomainIdentityMixin):
 
     __slots__ = (
         "id",

hcloud/hcloud.py

@@ -50,38 +50,41 @@ def _get_headers(self):
         }
         return headers
 
+    def _raise_exception_from_response(self, response):
+        raise HcloudAPIException(
+            code=response.status_code,
+            message=response.reason,
+            details={
+                'content': response.content
+            }
+        )
+
+    def _raise_exception_from_json_content(self, json_content):
+        raise HcloudAPIException(
+            code=json_content['error']['code'],
+            message=json_content['error']['message'],
+            details=json_content['error']['details']
+        )
+
     def request(self, method, url, **kwargs):
         response = requests.request(
             method,
             self.api_endpoint + url,
             headers=self._get_headers(),
             **kwargs
         )
-        result = response.content
+
+        json_content = response.content
         try:
-            if len(response.content) > 0:
-                result = response.json()
+            if len(json_content) > 0:
+                json_content = response.json()
         except (TypeError, ValueError):
-            raise HcloudAPIException(
-                code=response.status_code,
-                message=response.reason,
-                details={
-                    'content': response.content
-                }
-            )
+            self._raise_exception_from_response(response)
 
         if not response.ok:
-            if len(response.content) > 0:
-                raise HcloudAPIException(
-                    code=result['error']['code'],
-                    message=result['error']['message'],
-                    details=result['error']['details']
-                )
+            if json_content:
+                self._raise_exception_from_json_content(json_content)
             else:
-                raise HcloudAPIException(
-                    code="unknown_error",
-                    message="An unknown error occurred.",
-                    details=""
-                )
+                self._raise_exception_from_response(response)
 
-        return result
+        return json_content