feat(instance): add support set and get user_data

.github/workflows/checks.yml

@@ -104,7 +104,7 @@ jobs:
   #     - name: Set up Python
   #       uses: actions/setup-python@v5
   #       with:
-  #         python-version: 3.8
+  #         python-version: "3.10"
   #     - name: Install poetry
   #       run: |
   #         pip install poetry

scaleway-core/scaleway_core/api.py

@@ -122,10 +122,14 @@ def _request(
         if method == "POST" or method == "PUT" or method == "PATCH":
             additional_headers["Content-Type"] = "application/json; charset=utf-8"
 
-            if body is None:
-                body = {}
+        if body is None:
+            body = {}
 
-        raw_body = json.dumps(body) if body is not None else None
+        raw_body: Union[bytes, str]
+        if isinstance(body, bytes):
+            raw_body = body
+        else:
+            raw_body = json.dumps(body) if body is not None else None
 
         request_params: List[Tuple[str, Any]] = []
         for k, v in params.items():
@@ -155,9 +159,10 @@ def _request(
             url=url,
             params=request_params,
             headers=headers,
-            body=raw_body,
+            body=raw_body.decode("utf-8", errors="replace")
+            if isinstance(raw_body, bytes)
+            else raw_body,
         )
-
         response = requests.request(
             method=method,
             url=url,

scaleway/scaleway/instance/v1/custom_api.py

@@ -0,0 +1,130 @@
+from typing import Optional, Dict
+
+from requests import Response
+
+from scaleway_core.bridge import Zone as ScwZone
+from scaleway_core.utils import validate_path_param
+from .api import InstanceV1API
+from .custom_marshalling import marshal_GetServerUserDataRequest
+from .custom_types import GetServerUserDataRequest, GetAllServerUserDataResponse
+
+
+class InstanceUtilsV1API(InstanceV1API):
+    """
+    This API extends InstanceV1API by adding utility methods for managing Instance resources,
+    such as getting and setting server user data, while inheriting all methods of InstanceV1API.
+    """
+
+    def get_server_user_data(
+        self, server_id: str, key: str, zone: Optional[ScwZone] = None
+    ) -> Response:
+        """
+        GetServerUserData gets the content of a user data on a server for the given key.
+        :param zone: Zone to target. If none is passed will use default zone from the config.
+        :param server_id:
+        :param key:
+        :return: A plain text response with data user information
+
+        Usage:
+        ::
+
+            result = api.get_server_user_data(
+                server_id="example",
+                key="example",
+            )
+        """
+        param_zone = validate_path_param("zone", zone or self.client.default_zone)
+        param_server_id = validate_path_param("server_id", server_id)
+
+        res = self._request(
+            "GET",
+            f"/instance/v1/zones/{param_zone}/servers/{param_server_id}/user_data/{key}",
+            body=marshal_GetServerUserDataRequest(
+                GetServerUserDataRequest(
+                    zone=zone,
+                    server_id=server_id,
+                    key=key,
+                ),
+                self.client,
+            ),
+        )
+        self._throw_on_error(res)
+        return res
+
+    def set_server_user_data(
+        self, server_id: str, key: str, content: bytes, zone: Optional[ScwZone] = None
+    ) -> Response:
+        """
+        Sets the content of a user data on a server for the given key.
+        :param zone: Zone to target. If none is passed, it will use the default zone from the config.
+        :param server_id: The ID of the server.
+        :param key: The user data key.
+        :param content: The content to set as user data in bytes.
+        :return: A plain text response confirming the operation.
+        """
+        param_zone = validate_path_param("zone", zone or self.client.default_zone)
+        param_server_id = validate_path_param("server_id", server_id)
+        headers = {
+            "Content-Type": "text/plain",
+        }
+        res = self._request(
+            "PATCH",
+            f"/instance/v1/zones/{param_zone}/servers/{param_server_id}/user_data/{key}",
+            body=content,
+            headers=headers,
+        )
+
+        self._throw_on_error(res)
+        return res
+
+    def get_all_server_user_data(
+        self, server_id: str, zone: Optional[ScwZone] = None
+    ) -> GetAllServerUserDataResponse:
+        param_zone = validate_path_param("zone", zone or self.client.default_zone)
+        param_server_id = validate_path_param("server_id", server_id)
+
+        all_user_data_res = InstanceUtilsV1API.list_server_user_data(
+            self, server_id=param_server_id, zone=param_zone
+        )
+
+        user_data: Dict[str, bytes] = {}
+        for key in all_user_data_res.user_data:
+            value = InstanceUtilsV1API.get_server_user_data(
+                self, server_id=param_server_id, key=key
+            )
+            print("value: ", value)
+            user_data[key] = value.content
+
+        res = GetAllServerUserDataResponse(user_data=user_data)
+
+        return res
+
+    def set_all_server_user_data(
+        self,
+        server_id: str,
+        user_data: Dict[str, bytes],
+        zone: Optional[ScwZone] = None,
+    ) -> Optional[None]:
+        param_zone = validate_path_param("zone", zone or self.client.default_zone)
+        param_server_id = validate_path_param("server_id", server_id)
+
+        all_user_data_res = InstanceUtilsV1API.list_server_user_data(
+            self, server_id=param_server_id, zone=param_zone
+        )
+        for key in all_user_data_res.user_data:
+            if user_data.get(key) is not None:
+                continue
+            InstanceUtilsV1API.delete_server_user_data(
+                self, server_id=param_server_id, key=key
+            )
+
+        for key in user_data:
+            InstanceUtilsV1API.set_server_user_data(
+                self,
+                server_id=param_server_id,
+                zone=param_zone,
+                key=key,
+                content=user_data[key],
+            )
+
+        return None

scaleway/scaleway/instance/v1/custom_marshalling.py

@@ -0,0 +1,35 @@
+from typing import Dict, Any
+
+from scaleway.instance.v1.custom_types import (
+    GetServerUserDataRequest,
+    GetAllServerUserDataRequest,
+)
+from scaleway_core.profile import ProfileDefaults
+
+
+def marshal_GetServerUserDataRequest(
+    request: GetServerUserDataRequest, defaults: ProfileDefaults
+) -> Dict[str, Any]:
+    output: Dict[str, Any] = {}
+
+    if request.server_id is not None:
+        output["server_id"] = request.server_id
+    if request.key is not None:
+        output["key"] = request.key
+    if request.zone is not None:
+        output["zone"] = request.zone
+
+    return output
+
+
+def marshal_ListServerUserDataRequest(
+    request: GetAllServerUserDataRequest, defaults: ProfileDefaults
+) -> Dict[str, Any]:
+    output: Dict[str, Any] = {}
+
+    if request.server_id is not None:
+        output["server_id"] = request.server_id
+    if request.zone is not None:
+        output["zone"] = request.zone
+
+    return output

scaleway/scaleway/instance/v1/custom_types.py

@@ -0,0 +1,46 @@
+from dataclasses import dataclass
+from typing import Optional, Dict
+
+from scaleway_core.bridge import Zone as ScwZone
+
+
+@dataclass
+class GetServerUserDataRequest:
+    server_id: str
+
+    """
+    Key defines the user data key to get
+    """
+    key: str
+
+    """
+    Zone of the user data to get
+    """
+    zone: Optional[ScwZone]
+
+
+@dataclass
+class GetAllServerUserDataRequest:
+    server_id: str
+
+    """
+    Zone of the user data to get
+    """
+    zone: Optional[ScwZone]
+
+
+@dataclass
+class GetAllServerUserDataResponse:
+    user_data: Dict[str, bytes]
+
+
+@dataclass
+class SetAllServerUserDataRequest:
+    server_id: str
+
+    user_data: Dict[str, bytes]
+
+    """
+    Zone of the user data to set
+    """
+    zone: Optional[ScwZone]

scaleway/scaleway/instance/v1/test_user_data.py

@@ -0,0 +1,63 @@
+import sys
+import unittest
+import logging
+from typing import Dict
+
+from scaleway_core.client import Client
+from .custom_api import InstanceUtilsV1API
+
+logger = logging.getLogger()
+logger.level = logging.DEBUG
+stream_handler = logging.StreamHandler(sys.stdout)
+logger.addHandler(stream_handler)
+
+
+class TestServerUserData(unittest.TestCase):
+    def setUp(self) -> None:
+        self.client = Client()
+        self.instance_api = InstanceUtilsV1API(self.client, bypass_validation=True)
+        self.server = self.instance_api._create_server(
+            commercial_type="DEV1-S",
+            zone="fr-par-1",
+            image="ubuntu_jammy",
+            name="my-server-web",
+            volumes={},
+        )
+
+    @unittest.skip("API Test is not up")
+    def test_set_and_get_server_user_data(self) -> None:
+        if self.server is None or self.server.server is None:
+            self.fail("Server setup failed.")
+        key = "first key"
+        content = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x10"
+        self.instance_api.set_server_user_data(
+            server_id=self.server.server.id, key=key, content=content
+        )
+        user_data = self.instance_api.get_server_user_data(
+            server_id=self.server.server.id, key=key
+        )
+        self.assertIsNotNone(user_data)
+
+    @unittest.skip("API Test is not up")
+    def test_set_and_get_all_user_data(self) -> None:
+        if self.server is None or self.server.server is None:
+            self.fail("Server setup failed.")
+        key = "first key"
+        content = b"content first key"
+        key_bis = "second key"
+        content_bis = b"test content"
+        another_key = "third key"
+        another_content = b"another content to test"
+
+        user_data: Dict[str, bytes] = {
+            key_bis: content_bis,
+            another_key: another_content,
+            key: content,
+        }
+        self.instance_api.set_all_server_user_data(
+            server_id=self.server.server.id, user_data=user_data
+        )
+        response = self.instance_api.get_all_server_user_data(
+            server_id=self.server.server.id
+        )
+        self.assertIsNotNone(response)