fix: update workspaces

silverback/_cli.py

@@ -304,41 +304,40 @@ def new_workspace(
 @click.option(
     "-n",
     "--name",
-    "update_name",
+    "name",
     help="Update name for workspace",
 )
 @click.option(
     "-s",
     "--slug",
-    "update_slug",
+    "slug",
     help="Update slug for workspace",
 )
 @click.argument("workspace")
 @platform_client
 def update_workspace(
     platform: PlatformClient,
     workspace: str,
-    update_name: str | None,
-    update_slug: str | None,
+    name: str | None,
+    slug: str | None,
 ):
     """Update name and slug for a workspace"""
 
-    if not (platform.workspaces.get(workspace)):
+    if not (workspace_client := platform.workspaces.get(workspace)):
         raise click.BadOptionUsage("workspace", f"Unknown workspace '{workspace}'")
 
-    elif update_name is None and update_slug is None:
+    elif name is None and slug is None:
         raise click.UsageError(
             "No update name or slug found. Please enter a name or slug to update."
         )
-    elif update_name == "" or update_slug == "":
+    elif name == "" or slug == "":
         raise click.UsageError("Empty string value found for name or slug.")
 
-    platform.update_workspace(
-        workspace=workspace,
-        update_name=update_name,
-        update_slug=update_slug,
+    workspace = workspace_client.update(
+        name=name,
+        slug=slug,
     )
-    click.echo(f"{click.style('SUCCESS', fg='green')}: Updated '{workspace}'")
+    click.echo(f"{click.style('SUCCESS', fg='green')}: Updated '{workspace.name}'")
 
 
 @workspaces.command(name="delete", section="Platform Commands (https://silverback.apeworx.io)")
@@ -347,14 +346,14 @@ def update_workspace(
 def delete_workspace(platform: PlatformClient, workspace: str):
     """Delete an empty Workspace on the Silverback Platform"""
 
-    if not (workspace_client := platform.workspaces.get(workspace)):
+    if not (workspace := platform.workspaces.get(workspace)):
         raise click.BadOptionUsage("workspace", f"Unknown workspace '{workspace}'")
 
-    if len(workspace_client.clusters) > 0:
+    if len(workspace.clusters) > 0:
         raise click.UsageError("Running Clusters found in Workspace. Shut them down first.")
 
-    platform.remove_workspace(workspace)
-    click.echo(f"{click.style('SUCCESS', fg='green')}: Deleted '{workspace}'")
+    workspace.remove()
+    click.echo(f"{click.style('SUCCESS', fg='green')}: Deleted '{workspace.name}'")
 
 
 @cluster.command(name="list", section="Platform Commands (https://silverback.apeworx.io)")

silverback/cluster/client.py

@@ -402,6 +402,27 @@ def get_payment_stream(self, cluster: ClusterInfo, chain_id: int) -> Stream | No
 
         return Stream(manager=StreamManager(stream_info.manager), id=stream_info.stream_id)
 
+    def update(
+        self,
+        name: str | None = None,
+        slug: str | None = None,
+    ) -> "Workspace":
+        data = dict()
+        if name:
+            data["name"] = name
+        if slug:
+            data["slug"] = slug
+        response = self.client.patch(
+            f"/workspaces/{self.id}",
+            data=data,
+        )
+        handle_error_with_response(response)
+        return Workspace.model_validate_json(response.text)
+
+    def remove(self):
+        response = self.client.delete(f"/workspaces/{self.id}")
+        handle_error_with_response(response)
+
 
 class PlatformClient(httpx.Client):
     def __init__(self, *args, **kwargs):
@@ -452,32 +473,6 @@ def create_workspace(
         self.workspaces.update({new_workspace.slug: new_workspace})  # NOTE: Update cache
         return new_workspace
 
-    def remove_workspace(self, workspace_slug):
-        workspace_id = self.workspaces[workspace_slug].id
-        response = self.delete(f"/workspaces/{workspace_id}")
-        handle_error_with_response(response)
-
-    def update_workspace(
-        self,
-        workspace: str,
-        update_slug: str | None,
-        update_name: str | None,
-    ):
-        workspace_id = self.workspaces[workspace].id
-        form_data = dict()
-        if update_slug:
-            form_data["slug"] = update_slug
-        if update_name:
-            form_data["name"] = update_name
-        response = self.patch(
-            f"/workspaces/{workspace_id}",
-            data=form_data,
-        )
-        handle_error_with_response(response)
-        update_workspace = Workspace.model_validate_json(response.text)
-        self.workspaces.update({update_workspace.slug: update_workspace})  # NOTE: Update cache
-        return update_workspace
-
     def get_stream_manager(self, chain_id: int) -> StreamManager:
         response = self.get(f"/streams/manager/{chain_id}")
         handle_error_with_response(response)