Source code for linode_api4.groups.lke

from typing import Any, Dict, Optional, Union

from linode_api4.errors import UnexpectedResponseError
from linode_api4.groups import Group
from linode_api4.groups.lke_tier import LKETierGroup
from linode_api4.objects import (
    KubeVersion,
    LKECluster,
    LKEClusterControlPlaneOptions,
    LKEType,
    Type,
    drop_null_keys,
)
from linode_api4.objects.base import _flatten_request_body_recursive


[docs] class LKEGroup(Group): """ Encapsulates LKE-related methods of the :any:`LinodeClient`. This should not be instantiated on its own, but should instead be used through an instance of :any:`LinodeClient`:: client = LinodeClient(token) instances = client.lke.clusters() # use the LKEGroup This group contains all features beneath the `/lke` group in the API v4. """
[docs] def versions(self, *filters): """ Returns a :any:`PaginatedList` of :any:`KubeVersion` objects that can be used when creating an LKE Cluster. API Documentation: https://techdocs.akamai.com/linode-api/reference/get-lke-versions :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 KubeVersion """ return self.client._get_and_filter(KubeVersion, *filters)
[docs] def clusters(self, *filters): """ Returns a :any:`PaginagtedList` of :any:`LKECluster` objects that belong to this account. https://techdocs.akamai.com/linode-api/reference/get-lke-clusters :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 LKE clusters that match the query. :rtype: PaginatedList of LKECluster """ return self.client._get_and_filter(LKECluster, *filters)
[docs] def cluster_create( self, region, label, node_pools, kube_version, control_plane: Union[ LKEClusterControlPlaneOptions, Dict[str, Any] ] = None, apl_enabled: bool = False, tier: Optional[str] = None, **kwargs, ): """ Creates an :any:`LKECluster` on this account in the given region, with the given label, and with node pools as described. For example:: client = LinodeClient(TOKEN) # look up Region and Types to use. In this example I'm just using # the first ones returned. target_region = client.regions().first() node_type = client.linode.types()[0] node_type_2 = client.linode.types()[1] kube_version = client.lke.versions()[0] new_cluster = client.lke.cluster_create( target_region, "example-cluster", [client.lke.node_pool(node_type, 3), client.lke.node_pool(node_type_2, 3)], kube_version ) API Documentation: https://techdocs.akamai.com/linode-api/reference/post-lke-cluster :param region: The Region to create this LKE Cluster in. :type region: Region or str :param label: The label for the new LKE Cluster. :type label: str :param node_pools: The Node Pools to create. :type node_pools: one or a list of dicts containing keys "type" and "count". See :any:`node_pool` for a convenient way to create correctly- formatted dicts. :param kube_version: The version of Kubernetes to use :type kube_version: KubeVersion or str :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 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. :returns: The new LKE Cluster :rtype: LKECluster """ params = { "label": label, "region": region, "k8s_version": kube_version, "node_pools": ( node_pools if isinstance(node_pools, list) else [node_pools] ), "control_plane": control_plane, "tier": tier, } params.update(kwargs) # Prevent errors for users without access to APL if apl_enabled: params["apl_enabled"] = apl_enabled result = self.client.post( "/lke/clusters", data=drop_null_keys(_flatten_request_body_recursive(params)), ) if "id" not in result: raise UnexpectedResponseError( "Unexpected response when creating LKE cluster!", json=result ) return LKECluster(self.client, result["id"], result)
[docs] def node_pool(self, node_type: Union[Type, str], node_count: int, **kwargs): """ Returns a dict that is suitable for passing into the `node_pools` array of :any:`cluster_create`. This is a convenience method, and need not be used to create Node Pools. For proper usage, see the docs for :any:`cluster_create`. :param node_type: The type of node to create in this node pool. :type node_type: Type or str :param node_count: The number of nodes to create in this node pool. :type node_count: int :param kwargs: Other attributes to create this node pool with. :type kwargs: Any :returns: A dict describing the desired node pool. :rtype: dict """ result = { "type": node_type, "count": node_count, } result.update(kwargs) return result
[docs] def types(self, *filters): """ Returns a :any:`PaginatedList` of :any:`LKEType` objects that represents a valid LKE type. API Documentation: https://techdocs.akamai.com/linode-api/reference/get-lke-types :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 LKE types that match the query. :rtype: PaginatedList of LKEType """ return self.client._get_and_filter( LKEType, *filters, endpoint="/lke/types" )
[docs] def tier(self, id: str) -> LKETierGroup: """ Returns an object representing the LKE tier API path. NOTE: LKE tiers may not currently be available to all users. :param id: The ID of the tier. :type id: str :returns: An object representing the LKE tier API path. :rtype: LKETier """ return LKETierGroup(self.client, id)