WIP

Author: lgarber-akamai

linode_api4/__init__.py

@@ -1,5 +1,6 @@
 # isort: skip_file
 from linode_api4.objects import *
+from linode_api4.paths import *
 from linode_api4.errors import ApiError, UnexpectedResponseError
 from linode_api4.linode_client import LinodeClient
 from linode_api4.login_client import LinodeLoginClient, OAuthScopes

linode_api4/groups/lke.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict, Union
+from typing import Any, Dict, Optional, Union
 
 from linode_api4.errors import UnexpectedResponseError
 from linode_api4.groups import Group
@@ -11,6 +11,7 @@
     drop_null_keys,
 )
 from linode_api4.objects.base import _flatten_request_body_recursive
+from linode_api4.paths import LKETier
 
 
 class LKEGroup(Group):
@@ -67,6 +68,7 @@ def cluster_create(
             LKEClusterControlPlaneOptions, Dict[str, Any]
         ] = None,
         apl_enabled: bool = False,
+        tier: Optional[str] = None,
         **kwargs,
     ):
         """
@@ -104,9 +106,13 @@ def cluster_create(
         :param control_plane: The control plane configuration of this LKE cluster.
         :type control_plane: Dict[str, Any] or LKEClusterControlPlaneRequest
         :param apl_enabled: Whether this cluster should use APL.
-                            NOTE: This endpoint is in beta and may only
+                            NOTE: This field is in beta and may only
                             function if base_url is set to `https://api.linode.com/v4beta`.
         :type apl_enabled: bool
+        :param tier: The tier of LKE cluster to create.
+                     NOTE: This field is in beta and may only
+                     function if base_url is set to `https://api.linode.com/v4beta`.
+        :type tier: str
         :param kwargs: Any other arguments to pass along to the API.  See the API
                        docs for possible values.
 
@@ -122,6 +128,7 @@ def cluster_create(
                 node_pools if isinstance(node_pools, list) else [node_pools]
             ),
             "control_plane": control_plane,
+            "tier": tier,
         }
         params.update(kwargs)
 
@@ -183,3 +190,16 @@ def types(self, *filters):
         return self.client._get_and_filter(
             LKEType, *filters, endpoint="/lke/types"
         )
+
+    def tier(self, id: str) -> LKETier:
+        """
+        Returns an object representing the LKE tier API path.
+
+        :param id: The ID of the tier.
+        :type id: str
+
+        :returns: An object representing the LKE tier API path.
+        :rtype: LKETier
+        """
+
+        return LKETier(self.client, id)

linode_api4/objects/lke.py

@@ -49,6 +49,24 @@ class KubeVersion(Base):
     }
 
 
+class TieredKubeVersion(DerivedBase):
+    """
+    A TieredKubeVersion is a version of Kubernetes that is specific to a certain LKE tier.
+
+    API Documentation: https://techdocs.akamai.com/linode-api/reference/get-lke-version
+    """
+
+    api_endpoint = "/lke/tiers/{tier}/versions/{id}"
+    parent_id_name = "tier"
+    id_attribute = "id"
+    derived_url_path = "versions"
+
+    properties = {
+        "id": Property(identifier=True),
+        "tier": Property(identifier=True),
+    }
+
+
 @dataclass
 class LKENodePoolTaint(JSONObject):
     """
@@ -255,6 +273,7 @@ class LKECluster(Base):
         "pools": Property(derived_class=LKENodePool),
         "control_plane": Property(mutable=True),
         "apl_enabled": Property(),
+        "tier": Property(),
     }
 
     def invalidate(self):

linode_api4/paths/__init__.py

@@ -0,0 +1,3 @@
+# isort: skip_file
+from .api_path import *
+from .lke import *

linode_api4/paths/api_path.py

@@ -0,0 +1,15 @@
+from abc import ABC, abstractmethod
+
+
+class ApiPath(ABC):
+    """
+    ApiPath is an abstract class that represents a single static API path.
+    """
+
+    @property
+    @abstractmethod
+    def endpoint(self):
+        pass
+
+    def __repr__(self):
+        return f"{type(self).__name__}: {self.endpoint}"

linode_api4/paths/lke.py

@@ -0,0 +1,40 @@
+from linode_api4.objects.lke import TieredKubeVersion
+from linode_api4.paths.api_path import ApiPath
+
+
+class LKETier(ApiPath):
+    """
+    The API path for a single LKE tier.
+    """
+
+    def __init__(
+        self,
+        client: "LinodeClient",
+        id: str,
+    ):
+        self.client = client
+        self.id = id
+
+    @property
+    def endpoint(self) -> str:
+        return f"/lke/tiers/{self.id}"
+
+    def versions(self, *filters):
+        """
+        Returns a paginated list of versions for this tier matching the given filters.
+
+        API Documentation: Not Yet Available
+
+        :param filters: Any number of filters to apply to this query.
+                        See :doc:`Filtering Collections</linode_api4/objects/filtering>`
+                        for more details on filtering.
+
+        :returns: A paginated list of kube versions that match the query.
+        :rtype: PaginatedList of TieredKubeVersion
+        """
+
+        return self.client._get_and_filter(
+            TieredKubeVersion,
+            endpoint=f"{self.endpoint}/versions",
+            *filters,
+        )

test/fixtures/lke_tiers_standard_versions.json

@@ -0,0 +1,19 @@
+{
+  "data": [
+    {
+      "id": "1.32",
+      "tier": "standard"
+    },
+    {
+      "id": "1.31",
+      "tier": "standard"
+    },
+    {
+      "id": "1.30",
+      "tier": "standard"
+    }
+  ],
+  "page": 1,
+  "pages": 1,
+  "results": 3
+}

test/unit/objects/lke_test.py

@@ -2,7 +2,7 @@
 from test.unit.base import ClientBaseCase
 from unittest.mock import MagicMock
 
-from linode_api4 import InstanceDiskEncryptionType
+from linode_api4 import InstanceDiskEncryptionType, TieredKubeVersion
 from linode_api4.objects import (
     LKECluster,
     LKEClusterControlPlaneACLAddressesOptions,
@@ -536,3 +536,13 @@ def test_cluster_update_acl_null_addresses(self):
             # Addresses should not be included in the API request if it's null
             # See: TPT-3489
             assert m.call_data == {"acl": {"enabled": True}}
+
+    def test_lke_tiered_version(self):
+        version = TieredKubeVersion(self.client, "1.32", "standard")
+
+        assert version.id == "1.32"
+
+        # Ensure the version is properly refreshed
+        version.invalidate()
+
+        assert version.id == "1.32"

test/unit/paths/__init__.py

@@ -0,0 +1,19 @@
+from test.unit.base import ClientBaseCase
+
+
+
+class LKETierPathTest(ClientBaseCase):
+    """
+    Tests methods of the LKETierPath
+    """
+
+    def test_list_versions(self):
+        """
+        Tests that LKE versions can be listed for a given tier.
+        """
+
+        tiers = self.client.lke.tier("standard").versions()
+
+        assert tiers[0].id == "1.32"
+        assert tiers[1].id == "1.31"
+        assert tiers[2].id == "1.30"