Update Group Info.

src/tiledb/cloud/groups.py

@@ -1,14 +1,16 @@
 """Functions for managing TileDB Cloud groups."""
 
+import inspect
 import posixpath
 import urllib.parse
-from typing import Iterable, Optional, Tuple, Union
+from typing import Iterable, List, Optional, Tuple, Union
 
 import tiledb.cloud.tiledb_cloud_error as tce
 from tiledb.cloud import client
 from tiledb.cloud import rest_api
 from tiledb.cloud._common import api_v2
 from tiledb.cloud._common import utils
+from tiledb.cloud.rest_api.models import group_update
 
 
 def create(
@@ -92,6 +94,45 @@ def info(uri: str) -> object:
     return groups_client.get_group(group_namespace=namespace, group_name=group_name)
 
 
+def update_info(
+    uri: str,
+    *,
+    description: Optional[str] = None,
+    name: Optional[str] = None,
+    logo: Optional[str] = None,
+    tags: Optional[List[str]] = None,
+) -> None:
+    """
+    Update Group Metadata
+
+    :param uri: URI of the group in the form 'tiledb://<namespace>/<group>'
+    :param description: Group description, defaults to None
+    :param name: Group's name, defaults to None
+    :param logo: Group's logo, defaults to None
+    :param tags: Group tags, defaults to None
+    :return: None
+    """
+    namespace, group_name = utils.split_uri(uri)
+    groups_v1_client = client.build(rest_api.GroupsApi)
+    info = {}
+    for kw, arg in inspect.signature(update_info).parameters.items():
+        if arg.kind != inspect.Parameter.KEYWORD_ONLY:
+            # Skip every non-keyword-only argument
+            continue
+
+        value = locals()[kw]
+        if value is None:
+            # Explicitly update metadata
+            continue
+        info[kw] = value
+
+    info = group_update.GroupUpdate(**info)
+    try:
+        return groups_v1_client.update_group(namespace, group_name, group_update=info)
+    except rest_api.ApiException as exc:
+        raise tce.check_exc(exc)
+
+
 def deregister(
     uri: str,
     *,

tests/test_groups.py

@@ -15,6 +15,30 @@ def setUp(self):
         self.namespace, storage_path, _ = groups._default_ns_path_cred()
         self.test_path = storage_path + "/" + testonly.random_name("groups_test")
 
+    def assert_group_exists(self, name):
+        """Asserts that a group exists, and gives it a few tries."""
+        api = client.build(rest_api.GroupsApi)
+        for _ in range(TRIES):
+            try:
+                api.get_group_contents(group_namespace=self.namespace, group_name=name)
+                return
+            except rest_api.ApiException:
+                pass  # try again
+            time.sleep(1)
+        self.fail(f"group {self.namespace}/{name} does not exist")
+
+    def assert_group_not_exists(self, name):
+        """Asserts that a group does not exist, giving it a few tries to go."""
+        api = client.build(rest_api.GroupsApi)
+        for _ in range(TRIES):
+            try:
+                api.get_group_contents(group_namespace=self.namespace, group_name=name)
+            except rest_api.ApiException as apix:
+                if apix.status in (400, 404):
+                    return
+            time.sleep(1)
+        self.fail(f"group {self.namespace}/{name} still exists")
+
     def test_create_deregister(self):
         # TODO: This test leaves detritus around.
         # Once we get true delete functions, clean that up.
@@ -42,26 +66,24 @@ def test_create_deregister(self):
         self.assert_group_not_exists(outer_name)
         self.assert_group_not_exists(inner_name)
 
-    def assert_group_exists(self, name):
-        """Asserts that a group exists, and gives it a few tries."""
-        api = client.build(rest_api.GroupsApi)
-        for _ in range(TRIES):
-            try:
-                api.get_group_contents(group_namespace=self.namespace, group_name=name)
-                return
-            except rest_api.ApiException:
-                pass  # try again
-            time.sleep(1)
-        self.fail(f"group {self.namespace}/{name} does not exist")
+    def test_update_group_info(self):
+        group_name = testonly.random_name("test_group_info")
+        group_storage_name = testonly.random_name(group_name)
+        group_storage_path = f"{self.test_path}/{group_storage_name}"
+        group_uri = f"tiledb://{self.namespace}/{group_name}"
+        groups.create(group_name, storage_uri=group_storage_path)
+        self.assert_group_exists(group_name)
 
-    def assert_group_not_exists(self, name):
-        """Asserts that a group does not exist, giving it a few tries to go."""
-        api = client.build(rest_api.GroupsApi)
-        for _ in range(TRIES):
-            try:
-                api.get_group_contents(group_namespace=self.namespace, group_name=name)
-            except rest_api.ApiException as apix:
-                if apix.status in (400, 404):
-                    return
-            time.sleep(1)
-        self.fail(f"group {self.namespace}/{name} still exists")
+        description = "this is a test description"
+        logo = "testLogo"
+        tags = ["tag", "othertag"]
+        groups.update_info(group_uri, description=description, logo=logo, tags=tags)
+        group_info = groups.info(group_uri)
+
+        self.assertEqual(group_info.description, description)
+        self.assertEqual(group_info.logo, logo)
+        self.assertCountEqual(group_info.tags, tags)
+
+        # Cleanup
+        groups.deregister(group_uri)
+        self.assert_group_not_exists(group_name)