refactor: use typed namedtuples for the get_list page result (#251)

This is a nice refactor that remove some magic from the `get_list` methods.

The `get_list` function returns a named tuple that always has the following structure  `(result, meta)`. Since this is a tuple, we can unpack it without using the named attributes, and remove the `results_list_attribute_name` attributes used to access the data inside the tuple.

- Use predefined `NamedTuples` classes instead of dynamic (magic) `namedtuples` for the `get_list` `PageResult`. This will greatly improve the library typing definition.
- Remove unused functions and attributes after the refactor. This cleans up and simplify the code.

* refactor: simplify client _get_all implementation

* refactor: rewrite Meta.parse_meta method

* refactor: simplify GetEntityByNameMixin

* refactor: don't use dynamic page results namedtuple

* refactor: simplify ImagesClient.get_by_name_and_architecture

* refactor: remove unused results_list_attribute_name attr

* add comment about page result tuple unpacking

hcloud/actions/client.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
 import time
+from typing import NamedTuple
 
 from ..core.client import BoundModelBase, ClientEntityBase
+from ..core.domain import Meta
 from .domain import Action, ActionFailedException, ActionTimeoutException
 
 
@@ -29,9 +31,12 @@ def wait_until_finished(self, max_retries=100):
             raise ActionFailedException(action=self)
 
 
-class ActionsClient(ClientEntityBase):
-    results_list_attribute_name = "actions"
+class ActionsPageResult(NamedTuple):
+    actions: list[BoundAction]
+    meta: Meta | None
+
 
+class ActionsClient(ClientEntityBase):
     def get_by_id(self, id):
         # type: (int) -> BoundAction
         """Get a specific action by its ID.
@@ -77,7 +82,7 @@ def get_list(
         actions = [
             BoundAction(self, action_data) for action_data in response["actions"]
         ]
-        return self._add_meta_to_result(actions, response)
+        return ActionsPageResult(actions, Meta.parse_meta(response))
 
     def get_all(self, status=None, sort=None):
         # type: (Optional[List[str]], Optional[List[str]]) -> List[BoundAction]

hcloud/certificates/client.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
-from ..actions.client import BoundAction
+from typing import NamedTuple
+
+from ..actions.client import ActionsPageResult, BoundAction
 from ..core.client import BoundModelBase, ClientEntityBase, GetEntityByNameMixin
-from ..core.domain import add_meta_to_result
+from ..core.domain import Meta
 from .domain import (
     Certificate,
     CreateManagedCertificateResponse,
@@ -83,9 +85,12 @@ def retry_issuance(self):
         return self._client.retry_issuance(self)
 
 
-class CertificatesClient(ClientEntityBase, GetEntityByNameMixin):
-    results_list_attribute_name = "certificates"
+class CertificatesPageResult(NamedTuple):
+    certificates: list[BoundCertificate]
+    meta: Meta | None
 
+
+class CertificatesClient(ClientEntityBase, GetEntityByNameMixin):
     def get_by_id(self, id):
         # type: (int) -> BoundCertificate
         """Get a specific certificate by its ID.
@@ -138,7 +143,7 @@ def get_list(
             for certificate_data in response["certificates"]
         ]
 
-        return self._add_meta_to_result(certificates, response)
+        return CertificatesPageResult(certificates, Meta.parse_meta(response))
 
     def get_all(self, name=None, label_selector=None):
         # type: (Optional[str], Optional[str]) -> List[BoundCertificate]
@@ -287,7 +292,7 @@ def get_actions_list(
             BoundAction(self._client.actions, action_data)
             for action_data in response["actions"]
         ]
-        return add_meta_to_result(actions, response, "actions")
+        return ActionsPageResult(actions, Meta.parse_meta(response))
 
     def get_actions(self, certificate, status=None, sort=None):
         # type: (Certificate, Optional[List[str]], Optional[List[str]]) -> List[BoundAction]

hcloud/core/client.py

@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-from .domain import add_meta_to_result
-
 
 class ClientEntityBase:
     max_per_page = 50
@@ -14,44 +12,25 @@ 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 = []
 
+        page = 1
         while page:
-            page_result = list_function(
+            # The *PageResult tuples MUST have the following structure
+            # `(result: List[Bound*], meta: Meta)`
+            result, meta = 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
@@ -66,17 +45,14 @@ def _get_all(
 
     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
-        )
+        return self._get_all(self.get_list, *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)
+        return self._get_all(self.get_actions_list, *args, **kwargs)
 
 
 class GetEntityByNameMixin:
@@ -86,11 +62,8 @@ class GetEntityByNameMixin:
 
     def get_by_name(self, name):
         # type: (str) -> BoundModelBase
-        self._is_list_attribute_implemented()
-        response = self.get_list(name=name)
-        entities = getattr(response, self.results_list_attribute_name)
-        entity = entities[0] if entities else None
-        return entity
+        entities, _ = self.get_list(name=name)
+        return entities[0] if entities else None
 
 
 class BoundModelBase:

hcloud/core/domain.py

@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-from collections import namedtuple
-
 
 class BaseDomain:
     __slots__ = ()
@@ -63,19 +61,13 @@ def __init__(self, pagination=None):
         self.pagination = pagination
 
     @classmethod
-    def parse_meta(cls, json_content):
+    def parse_meta(cls, response: dict) -> Meta | None:
         meta = None
-        if json_content and "meta" in json_content:
+        if response and "meta" in response:
             meta = cls()
-            pagination_json = json_content["meta"].get("pagination")
-            if pagination_json:
-                pagination = Pagination(**pagination_json)
-                meta.pagination = pagination
-        return meta
+            try:
+                meta.pagination = Pagination(**response["meta"]["pagination"])
+            except KeyError:
+                pass
 
-
-def add_meta_to_result(result, json_content, attr_name):
-    # type: (List[BoundModelBase], json, string) -> PageResult
-    class_name = f"PageResults{attr_name.capitalize()}"
-    PageResults = namedtuple(class_name, [attr_name, "meta"])
-    return PageResults(**{attr_name: result, "meta": Meta.parse_meta(json_content)})
+        return meta

hcloud/datacenters/client.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
+from typing import NamedTuple
+
 from ..core.client import BoundModelBase, ClientEntityBase, GetEntityByNameMixin
+from ..core.domain import Meta
 from ..locations.client import BoundLocation
 from ..server_types.client import BoundServerType
 from .domain import Datacenter, DatacenterServerTypes
@@ -43,9 +46,12 @@ def __init__(self, client, data):
         super().__init__(client, data)
 
 
-class DatacentersClient(ClientEntityBase, GetEntityByNameMixin):
-    results_list_attribute_name = "datacenters"
+class DatacentersPageResult(NamedTuple):
+    datacenters: list[BoundDatacenter]
+    meta: Meta | None
 
+
+class DatacentersClient(ClientEntityBase, GetEntityByNameMixin):
     def get_by_id(self, id):
         # type: (int) -> BoundDatacenter
         """Get a specific datacenter by its ID.
@@ -90,7 +96,7 @@ def get_list(
             for datacenter_data in response["datacenters"]
         ]
 
-        return self._add_meta_to_result(datacenters, response)
+        return DatacentersPageResult(datacenters, Meta.parse_meta(response))
 
     def get_all(self, name=None):
         # type: (Optional[str]) -> List[BoundDatacenter]

hcloud/firewalls/client.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
-from ..actions.client import BoundAction
+from typing import NamedTuple
+
+from ..actions.client import ActionsPageResult, BoundAction
 from ..core.client import BoundModelBase, ClientEntityBase, GetEntityByNameMixin
-from ..core.domain import add_meta_to_result
+from ..core.domain import Meta
 from .domain import (
     CreateFirewallResponse,
     Firewall,
@@ -134,9 +136,12 @@ def remove_from_resources(self, resources):
         return self._client.remove_from_resources(self, resources)
 
 
-class FirewallsClient(ClientEntityBase, GetEntityByNameMixin):
-    results_list_attribute_name = "firewalls"
+class FirewallsPageResult(NamedTuple):
+    firewalls: list[BoundFirewall]
+    meta: Meta | None
 
+
+class FirewallsClient(ClientEntityBase, GetEntityByNameMixin):
     def get_actions_list(
         self,
         firewall,  # type: Firewall
@@ -177,7 +182,7 @@ def get_actions_list(
             BoundAction(self._client.actions, action_data)
             for action_data in response["actions"]
         ]
-        return add_meta_to_result(actions, response, "actions")
+        return ActionsPageResult(actions, Meta.parse_meta(response))
 
     def get_actions(
         self,
@@ -249,7 +254,7 @@ def get_list(
             for firewall_data in response["firewalls"]
         ]
 
-        return self._add_meta_to_result(firewalls, response)
+        return FirewallsPageResult(firewalls, Meta.parse_meta(response))
 
     def get_all(self, label_selector=None, name=None, sort=None):
         # type: (Optional[str], Optional[str],  Optional[List[str]]) -> List[BoundFirewall]

hcloud/floating_ips/client.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
-from ..actions.client import BoundAction
+from typing import NamedTuple
+
+from ..actions.client import ActionsPageResult, BoundAction
 from ..core.client import BoundModelBase, ClientEntityBase, GetEntityByNameMixin
-from ..core.domain import add_meta_to_result
+from ..core.domain import Meta
 from ..locations.client import BoundLocation
 from .domain import CreateFloatingIPResponse, FloatingIP
 
@@ -119,9 +121,12 @@ def change_dns_ptr(self, ip, dns_ptr):
         return self._client.change_dns_ptr(self, ip, dns_ptr)
 
 
-class FloatingIPsClient(ClientEntityBase, GetEntityByNameMixin):
-    results_list_attribute_name = "floating_ips"
+class FloatingIPsPageResult(NamedTuple):
+    floating_ips: list[BoundFloatingIP]
+    meta: Meta | None
 
+
+class FloatingIPsClient(ClientEntityBase, GetEntityByNameMixin):
     def get_actions_list(
         self,
         floating_ip,  # type: FloatingIP
@@ -164,7 +169,7 @@ def get_actions_list(
             BoundAction(self._client.actions, action_data)
             for action_data in response["actions"]
         ]
-        return add_meta_to_result(actions, response, "actions")
+        return ActionsPageResult(actions, Meta.parse_meta(response))
 
     def get_actions(
         self,
@@ -234,7 +239,7 @@ def get_list(
             for floating_ip_data in response["floating_ips"]
         ]
 
-        return self._add_meta_to_result(floating_ips, response)
+        return FloatingIPsPageResult(floating_ips, Meta.parse_meta(response))
 
     def get_all(self, label_selector=None, name=None):
         # type: (Optional[str], Optional[str]) -> List[BoundFloatingIP]

hcloud/images/client.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
-from ..actions.client import BoundAction
+from typing import NamedTuple
+
+from ..actions.client import ActionsPageResult, BoundAction
 from ..core.client import BoundModelBase, ClientEntityBase, GetEntityByNameMixin
-from ..core.domain import add_meta_to_result
+from ..core.domain import Meta
 from .domain import Image
 
 
@@ -89,9 +91,12 @@ def change_protection(self, delete=None):
         return self._client.change_protection(self, delete)
 
 
-class ImagesClient(ClientEntityBase, GetEntityByNameMixin):
-    results_list_attribute_name = "images"
+class ImagesPageResult(NamedTuple):
+    images: list[BoundImage]
+    meta: Meta | None
 
+
+class ImagesClient(ClientEntityBase, GetEntityByNameMixin):
     def get_actions_list(
         self,
         image,  # type: Image
@@ -132,7 +137,7 @@ def get_actions_list(
             BoundAction(self._client.actions, action_data)
             for action_data in response["actions"]
         ]
-        return add_meta_to_result(actions, response, "actions")
+        return ActionsPageResult(actions, Meta.parse_meta(response))
 
     def get_actions(
         self,
@@ -224,7 +229,7 @@ def get_list(
         response = self._client.request(url="/images", method="GET", params=params)
         images = [BoundImage(self, image_data) for image_data in response["images"]]
 
-        return self._add_meta_to_result(images, response)
+        return ImagesPageResult(images, Meta.parse_meta(response))
 
     def get_all(
         self,
@@ -291,8 +296,7 @@ def get_by_name_and_architecture(self, name, architecture):
                Used to identify the image.
         :return: :class:`BoundImage <hcloud.images.client.BoundImage>`
         """
-        response = self.get_list(name=name, architecture=[architecture])
-        entities = getattr(response, self.results_list_attribute_name)
+        entities, _ = self.get_list(name=name, architecture=[architecture])
         entity = entities[0] if entities else None
         return entity