refactor: split top level client from base client

hcloud/_client.py

@@ -78,6 +78,25 @@ def func(retries: int) -> float:
     return func
 
 
+def _build_user_agent(
+    application_name: str | None,
+    application_version: str | None,
+) -> str:
+    """Build the user agent of the hcloud-python instance with the user application name (if specified)
+
+    :return: The user agent of this hcloud-python instance
+    """
+    parts = []
+    for name, version in [
+        (application_name, application_version),
+        ("hcloud-python", __version__),
+    ]:
+        if name is not None:
+            parts.append(name if version is None else f"{name}/{version}")
+
+    return " ".join(parts)
+
+
 class Client:
     """
     Client for the Hetzner Cloud API.
@@ -112,14 +131,6 @@ class Client:
     breaking changes.
     """
 
-    _version = __version__
-    __user_agent_prefix = "hcloud-python"
-
-    _retry_interval = staticmethod(
-        exponential_backoff_function(base=1.0, multiplier=2, cap=60.0, jitter=True)
-    )
-    _retry_max_retries = 5
-
     def __init__(
         self,
         token: str,
@@ -143,18 +154,15 @@ def __init__(
             Max retries before timeout when polling actions from the API.
         :param timeout: Requests timeout 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._requests_timeout = timeout
-
-        if isinstance(poll_interval, (int, float)):
-            self._poll_interval_func = constant_backoff_function(poll_interval)
-        else:
-            self._poll_interval_func = poll_interval
-        self._poll_max_retries = poll_max_retries
+        self._client = ClientBase(
+            token=token,
+            endpoint=api_endpoint,
+            application_name=application_name,
+            application_version=application_version,
+            poll_interval=poll_interval,
+            poll_max_retries=poll_max_retries,
+            timeout=timeout,
+        )
 
         self.datacenters = DatacentersClient(self)
         """DatacentersClient Instance
@@ -246,50 +254,81 @@ def __init__(
         :type: :class:`PlacementGroupsClient <hcloud.placement_groups.client.PlacementGroupsClient>`
         """
 
-    def _get_user_agent(self) -> str:
-        """Get the user agent of the hcloud-python instance with the user application name (if specified)
+    def request(  # type: ignore[no-untyped-def]
+        self,
+        method: str,
+        url: str,
+        **kwargs,
+    ) -> dict:
+        """Perform a request to the Hetzner Cloud API.
 
-        :return: The user agent of this hcloud-python instance
+        :param method: Method to perform the request.
+        :param url: URL to perform the request.
+        :param timeout: Requests timeout in seconds.
         """
-        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 self._client.request(method, url, **kwargs)
+
+
+class ClientBase:
+    def __init__(
+        self,
+        token: str,
+        *,
+        endpoint: str,
+        application_name: str | None = None,
+        application_version: str | None = None,
+        poll_interval: int | float | BackoffFunction = 1.0,
+        poll_max_retries: int = 120,
+        timeout: float | tuple[float, float] | None = None,
+    ):
+        self._token = token
+        self._endpoint = endpoint
+
+        self._user_agent = _build_user_agent(application_name, application_version)
+        self._headers = {
+            "User-Agent": self._user_agent,
+            "Authorization": f"Bearer {self._token}",
+            "Accept": "application/json",
         }
-        return headers
+
+        if isinstance(poll_interval, (int, float)):
+            poll_interval_func = constant_backoff_function(poll_interval)
+        else:
+            poll_interval_func = poll_interval
+
+        self._poll_interval_func = poll_interval_func
+        self._poll_max_retries = poll_max_retries
+
+        self._retry_interval_func = exponential_backoff_function(
+            base=1.0, multiplier=2, cap=60.0, jitter=True
+        )
+        self._retry_max_retries = 5
+
+        self._timeout = timeout
+        self._session = requests.Session()
 
     def request(  # type: ignore[no-untyped-def]
         self,
         method: str,
         url: str,
         **kwargs,
     ) -> dict:
-        """Perform a request to the Hetzner Cloud API, wrapper around requests.request
+        """Perform a request to the provided URL.
 
-        :param method: HTTP Method to perform the Request
-        :param url: URL of the Endpoint
-        :param timeout: Requests timeout in seconds
+        :param method: Method to perform the request.
+        :param url: URL to perform the request.
+        :param timeout: Requests timeout in seconds.
         :return: Response
         """
-        kwargs.setdefault("timeout", self._requests_timeout)
+        kwargs.setdefault("timeout", self._timeout)
 
-        url = self._api_endpoint + url
-        headers = self._get_headers()
+        url = self._endpoint + url
+        headers = self._headers
 
         retries = 0
         while True:
             try:
-                response = self._requests_session.request(
+                response = self._session.request(
                     method=method,
                     url=url,
                     headers=headers,
@@ -298,13 +337,13 @@ def request(  # type: ignore[no-untyped-def]
                 return self._read_response(response)
             except APIException as exception:
                 if retries < self._retry_max_retries and self._retry_policy(exception):
-                    time.sleep(self._retry_interval(retries))
+                    time.sleep(self._retry_interval_func(retries))
                     retries += 1
                     continue
                 raise
             except requests.exceptions.Timeout:
                 if retries < self._retry_max_retries:
-                    time.sleep(self._retry_interval(retries))
+                    time.sleep(self._retry_interval_func(retries))
                     retries += 1
                     continue
                 raise

hcloud/actions/client.py

@@ -53,8 +53,16 @@ class ActionsPageResult(NamedTuple):
 class ResourceActionsClient(ResourceClientBase):
     _resource: str
 
-    def __init__(self, client: Client, resource: str | None):
-        super().__init__(client)
+    def __init__(self, client: ResourceClientBase | Client, resource: str | None):
+        if isinstance(client, ResourceClientBase):
+            super().__init__(client._parent)
+            # Use the same base client as the the resource base client. Allows us to
+            # choose the base client outside of the ResourceActionsClient.
+            self._client = client._client
+        else:
+            # Backward compatibility, defaults to the parent ("top level") base client (`_client`).
+            super().__init__(client)
+
         self._resource = resource or ""
 
     def get_by_id(self, id: int) -> BoundAction:
@@ -67,7 +75,7 @@ def get_by_id(self, id: int) -> BoundAction:
             url=f"{self._resource}/actions/{id}",
             method="GET",
         )
-        return BoundAction(self._client.actions, response["action"])
+        return BoundAction(self._parent.actions, response["action"])
 
     def get_list(
         self,
@@ -104,7 +112,7 @@ def get_list(
             params=params,
         )
         actions = [
-            BoundAction(self._client.actions, action_data)
+            BoundAction(self._parent.actions, action_data)
             for action_data in response["actions"]
         ]
         return ActionsPageResult(actions, Meta.parse_meta(response))

hcloud/certificates/client.py

@@ -104,7 +104,6 @@ class CertificatesPageResult(NamedTuple):
 
 
 class CertificatesClient(ResourceClientBase):
-    _client: Client
 
     actions: ResourceActionsClient
     """Certificates scoped actions client
@@ -248,7 +247,7 @@ def create_managed(
         response = self._client.request(url="/certificates", method="POST", json=data)
         return CreateManagedCertificateResponse(
             certificate=BoundCertificate(self, response["certificate"]),
-            action=BoundAction(self._client.actions, response["action"]),
+            action=BoundAction(self._parent.actions, response["action"]),
         )
 
     def update(
@@ -328,7 +327,7 @@ def get_actions_list(
             params=params,
         )
         actions = [
-            BoundAction(self._client.actions, action_data)
+            BoundAction(self._parent.actions, action_data)
             for action_data in response["actions"]
         ]
         return ActionsPageResult(actions, Meta.parse_meta(response))
@@ -368,4 +367,4 @@ def retry_issuance(
             url=f"/certificates/{certificate.id}/actions/retry",
             method="POST",
         )
-        return BoundAction(self._client.actions, response["action"])
+        return BoundAction(self._parent.actions, response["action"])

hcloud/core/client.py

@@ -4,21 +4,20 @@
 from typing import TYPE_CHECKING, Any, Callable
 
 if TYPE_CHECKING:
-    from .._client import Client
+    from .._client import Client, ClientBase
     from .domain import BaseDomain
 
 
 class ResourceClientBase:
-    _client: Client
+    _parent: Client
+    _client: ClientBase
 
     max_per_page: int = 50
 
     def __init__(self, client: Client):
-        """
-        :param client: Client
-        :return self
-        """
-        self._client = client
+        self._parent = client
+        # Use the parent "default" base client.
+        self._client = client._client
 
     def _iter_pages(  # type: ignore[no-untyped-def]
         self,

hcloud/datacenters/client.py

@@ -1,15 +1,12 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any, NamedTuple
+from typing import Any, NamedTuple
 
 from ..core import BoundModelBase, Meta, ResourceClientBase
 from ..locations import BoundLocation
 from ..server_types import BoundServerType
 from .domain import Datacenter, DatacenterServerTypes
 
-if TYPE_CHECKING:
-    from .._client import Client
-
 
 class BoundDatacenter(BoundModelBase, Datacenter):
     _client: DatacentersClient
@@ -19,25 +16,25 @@ class BoundDatacenter(BoundModelBase, Datacenter):
     def __init__(self, client: DatacentersClient, data: dict):
         location = data.get("location")
         if location is not None:
-            data["location"] = BoundLocation(client._client.locations, location)
+            data["location"] = BoundLocation(client._parent.locations, location)
 
         server_types = data.get("server_types")
         if server_types is not None:
             available = [
                 BoundServerType(
-                    client._client.server_types, {"id": server_type}, complete=False
+                    client._parent.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
+                    client._parent.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
+                    client._parent.server_types, {"id": server_type}, complete=False
                 )
                 for server_type in server_types["available_for_migration"]
             ]
@@ -56,7 +53,6 @@ class DatacentersPageResult(NamedTuple):
 
 
 class DatacentersClient(ResourceClientBase):
-    _client: Client
 
     def get_by_id(self, id: int) -> BoundDatacenter:
         """Get a specific datacenter by its ID.