feat(k8s): expose MigrateClusterToSBSCSI

scaleway-async/scaleway_async/k8s/v1/__init__.py

@@ -69,6 +69,7 @@
 from .types import ListVersionsRequest
 from .types import ListVersionsResponse
 from .types import MigrateClusterToRoutedIPsRequest
+from .types import MigrateClusterToSBSCSIRequest
 from .types import NodeMetadata
 from .types import RebootNodeRequest
 from .types import ReplaceNodeRequest
@@ -150,6 +151,7 @@
     "ListVersionsRequest",
     "ListVersionsResponse",
     "MigrateClusterToRoutedIPsRequest",
+    "MigrateClusterToSBSCSIRequest",
     "NodeMetadata",
     "RebootNodeRequest",
     "ReplaceNodeRequest",

scaleway-async/scaleway_async/k8s/v1/api.py

@@ -749,6 +749,41 @@ async def migrate_cluster_to_routed_i_ps(
         self._throw_on_error(res)
         return unmarshal_Cluster(res.json())
 
+    async def migrate_cluster_to_sbscsi(
+        self,
+        *,
+        cluster_id: str,
+        region: Optional[Region] = None,
+    ) -> Cluster:
+        """
+        Migrate a cluster to SBS CSI.
+        Enable the latest CSI compatible with Scaleway Block Storage (SBS) and migrate all existing PersistentVolumes/VolumeSnapshotContents to SBS.
+        :param cluster_id: Cluster ID for which the latest CSI compatible with Scaleway Block Storage will be enabled.
+        :param region: Region to target. If none is passed will use default region from the config.
+        :return: :class:`Cluster <Cluster>`
+
+        Usage:
+        ::
+
+            result = await api.migrate_cluster_to_sbscsi(
+                cluster_id="example",
+            )
+        """
+
+        param_region = validate_path_param(
+            "region", region or self.client.default_region
+        )
+        param_cluster_id = validate_path_param("cluster_id", cluster_id)
+
+        res = self._request(
+            "POST",
+            f"/k8s/v1/regions/{param_region}/clusters/{param_cluster_id}/migrate-to-sbs-csi",
+            body={},
+        )
+
+        self._throw_on_error(res)
+        return unmarshal_Cluster(res.json())
+
     async def list_pools(
         self,
         *,

scaleway-async/scaleway_async/k8s/v1/types.py

@@ -1707,6 +1707,19 @@ class MigrateClusterToRoutedIPsRequest:
     """
 
 
+@dataclass
+class MigrateClusterToSBSCSIRequest:
+    cluster_id: str
+    """
+    Cluster ID for which the latest CSI compatible with Scaleway Block Storage will be enabled.
+    """
+
+    region: Optional[Region]
+    """
+    Region to target. If none is passed will use default region from the config.
+    """
+
+
 @dataclass
 class NodeMetadata:
     id: str

scaleway/scaleway/k8s/v1/__init__.py

@@ -69,6 +69,7 @@
 from .types import ListVersionsRequest
 from .types import ListVersionsResponse
 from .types import MigrateClusterToRoutedIPsRequest
+from .types import MigrateClusterToSBSCSIRequest
 from .types import NodeMetadata
 from .types import RebootNodeRequest
 from .types import ReplaceNodeRequest
@@ -150,6 +151,7 @@
     "ListVersionsRequest",
     "ListVersionsResponse",
     "MigrateClusterToRoutedIPsRequest",
+    "MigrateClusterToSBSCSIRequest",
     "NodeMetadata",
     "RebootNodeRequest",
     "ReplaceNodeRequest",

scaleway/scaleway/k8s/v1/api.py

@@ -749,6 +749,41 @@ def migrate_cluster_to_routed_i_ps(
         self._throw_on_error(res)
         return unmarshal_Cluster(res.json())
 
+    def migrate_cluster_to_sbscsi(
+        self,
+        *,
+        cluster_id: str,
+        region: Optional[Region] = None,
+    ) -> Cluster:
+        """
+        Migrate a cluster to SBS CSI.
+        Enable the latest CSI compatible with Scaleway Block Storage (SBS) and migrate all existing PersistentVolumes/VolumeSnapshotContents to SBS.
+        :param cluster_id: Cluster ID for which the latest CSI compatible with Scaleway Block Storage will be enabled.
+        :param region: Region to target. If none is passed will use default region from the config.
+        :return: :class:`Cluster <Cluster>`
+
+        Usage:
+        ::
+
+            result = api.migrate_cluster_to_sbscsi(
+                cluster_id="example",
+            )
+        """
+
+        param_region = validate_path_param(
+            "region", region or self.client.default_region
+        )
+        param_cluster_id = validate_path_param("cluster_id", cluster_id)
+
+        res = self._request(
+            "POST",
+            f"/k8s/v1/regions/{param_region}/clusters/{param_cluster_id}/migrate-to-sbs-csi",
+            body={},
+        )
+
+        self._throw_on_error(res)
+        return unmarshal_Cluster(res.json())
+
     def list_pools(
         self,
         *,

scaleway/scaleway/k8s/v1/types.py

@@ -1707,6 +1707,19 @@ class MigrateClusterToRoutedIPsRequest:
     """
 
 
+@dataclass
+class MigrateClusterToSBSCSIRequest:
+    cluster_id: str
+    """
+    Cluster ID for which the latest CSI compatible with Scaleway Block Storage will be enabled.
+    """
+
+    region: Optional[Region]
+    """
+    Region to target. If none is passed will use default region from the config.
+    """
+
+
 @dataclass
 class NodeMetadata:
     id: str