Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: add workspace subcommands #148

Merged
merged 17 commits into from
Oct 25, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
121 changes: 119 additions & 2 deletions silverback/_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -224,9 +224,14 @@ def cluster():
your platform account via `-c WORKSPACE/NAME`"""


@cluster.command(section="Platform Commands (https://silverback.apeworx.io)")
@cluster.group(cls=SectionedHelpGroup, section="Platform Commands (https://silverback.apeworx.io)")
def workspaces():
"""View and Manage Workspaces on the Silverback Platform"""


@workspaces.command(name="list", section="Platform Commands (https://silverback.apeworx.io)")
@platform_client
def workspaces(platform: PlatformClient):
def list_workspaces(platform: PlatformClient):
"""List available workspaces for your account"""

if workspace_names := list(platform.workspaces):
Expand All @@ -241,6 +246,115 @@ def workspaces(platform: PlatformClient):
)


@workspaces.command(name="info", section="Platform Commands (https://silverback.apeworx.io)")
@click.argument("workspace")
@platform_client
def workspace_info(platform: PlatformClient, workspace: str):
"""Get Configuration information about a WORKSPACE"""

if not (workspace_info := platform.workspaces.get(workspace)):
raise click.BadOptionUsage("workspace", f"Unknown workspace '{workspace}'")

click.echo(f"{click.style('Name', fg='green')}: {workspace_info.name}")
click.echo(f"{click.style('Slug', fg='green')}: '{workspace_info.slug}'")
click.echo(f"{click.style('Date Created', fg='green')}: '{workspace_info.created}'")


@workspaces.command(name="new", section="Platform Commands (https://silverback.apeworx.io)")
@click.option(
"-n",
"--name",
"workspace_name",
required=True,
help="Name for new workspace",
)
@click.option(
"-s",
"--slug",
"workspace_slug",
required=True,
help="Slug for new workspace",
)
@platform_client
def new_workspace(
platform: PlatformClient,
workspace_name: str,
workspace_slug: str,
):
"""Create a new workspace"""

if workspace_name:
click.echo(f"name: {workspace_name}")
click.echo(f"slug: {workspace_slug or workspace_name.lower().replace(' ', '-')}")

elif workspace_slug:
click.echo(f"slug: {workspace_slug}")

dtdang marked this conversation as resolved.
Show resolved Hide resolved
else:
raise click.UsageError("Must provide a name or a slug/name combo")

platform.create_workspace(
dtdang marked this conversation as resolved.
Show resolved Hide resolved
workspace_name=workspace_name,
workspace_slug=workspace_slug,
)
click.echo(f"{click.style('SUCCESS', fg='green')}: Created '{workspace_name}'")

dtdang marked this conversation as resolved.
Show resolved Hide resolved

@workspaces.command(name="update", section="Platform Commands (https://silverback.apeworx.io)")
@click.option(
"-n",
"--name",
"update_name",
help="Update name for workspace",
)
@click.option(
"-s",
"--slug",
"update_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,
):
"""Update name and slug for a workspace"""

if not (platform.workspaces.get(workspace)):
raise click.BadOptionUsage("workspace", f"Unknown workspace '{workspace}'")

elif update_name is None and update_slug is None:
raise click.UsageError(
"No update name or slug found. Please enter a name or slug to update."
)

platform.update_workspace(
workspace=workspace,
update_name=update_name,
update_slug=update_slug,
)
click.echo(f"{click.style('SUCCESS', fg='green')}: Updated '{update_name}'")


@workspaces.command(name="delete", section="Platform Commands (https://silverback.apeworx.io)")
@click.argument("workspace")
@platform_client
def delete_workspace(platform: PlatformClient, workspace: str):
dtdang marked this conversation as resolved.
Show resolved Hide resolved
"""Delete an empty Workspace on the Silverback Platform"""

if not (workspace_client := platform.workspaces.get(workspace)):
raise click.BadOptionUsage("workspace", f"Unknown workspace '{workspace}'")

if len(workspace_client.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}'")


@cluster.command(name="list", section="Platform Commands (https://silverback.apeworx.io)")
@click.argument("workspace")
@platform_client
Expand Down Expand Up @@ -290,6 +404,9 @@ def new_cluster(
elif cluster_slug:
click.echo(f"slug: {cluster_slug}")

else:
raise click.UsageError("Must provide a name or a slug/name combo")

cluster = workspace_client.create_cluster(
cluster_name=cluster_name,
cluster_slug=cluster_slug,
Expand Down
21 changes: 21 additions & 0 deletions silverback/cluster/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -452,6 +452,27 @@ 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
response = self.patch(
f"/workspaces/{workspace_id}",
data=dict(slug=update_slug, name=update_name),
)
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)
Expand Down
Loading