feat: Add --workspace-id flag to latch register

src/latch_cli/main.py

@@ -656,6 +656,16 @@ def execute(
         "the package root if one exists, or (2) generate one in .latch/Dockerfile if none exists."
     ),
 )
+@click.option(
+    "--workspace-id",
+    type=str,
+    default=None,
+    help=(
+        "Register the workflow to the specified workspace. "
+        "This argument accepts a numeric workspace ID, e.g. `--workspace-id 1234`. "
+        "By default, workflows are registered to the active workspace."
+    ),
+)
 @requires_login
 def register(
     pkg_root: str,
@@ -673,6 +683,7 @@ def register(
     mark_as_release: bool,
     staging: bool,
     dockerfile: Optional[Path],
+    workspace_id: Optional[str],
 ):
     """Register local workflow code to Latch.
 
@@ -713,6 +724,14 @@ def register(
         )
         raise click.exceptions.Exit(1)
 
+    if workspace_id is not None and not workspace_id.isdigit():
+        click.secho(
+            f"The value provided for --workspace-id must be an integer: {workspace_id}",
+            fg="red",
+            bold=True,
+        )
+        raise click.exceptions.Exit(1)
+
     from latch_cli.services.register import register
 
     register(
@@ -732,6 +751,7 @@ def register(
         cache_tasks=cache_tasks,
         mark_as_release=mark_as_release,
         dockerfile_path=dockerfile,
+        workspace_id=workspace_id,
     )
 
 

src/latch_cli/services/register/register.py

@@ -305,6 +305,7 @@ def register(
     use_new_centromere: bool = False,
     mark_as_release: bool = False,
     dockerfile_path: Optional[Path] = None,
+    workspace_id: Optional[str] = None,
 ):
     """Registers a workflow, defined as python code, with Latch.
 
@@ -382,16 +383,27 @@ def register(
         )
         click.echo(" ".join([click.style("Version:", fg="bright_blue"), ctx.version]))
 
+        if workspace_id is None:
+            workspace_id = current_workspace()
         workspaces = get_workspaces()
+
+        if workspace_id not in workspaces:
+            click.secho(
+                f"User does not have permission to access workspace {workspace_id}.",
+                fg="red",
+                bold=True,
+            )
+            raise click.exceptions.Exit(1)
+
         ws_name = next(
-            (x[1]["name"] for x in workspaces.items() if x[0] == current_workspace()),
+            (x[1]["name"] for x in workspaces.items() if x[0] == workspace_id),
             "N/A",
         )
         click.echo(
             " ".join([
                 click.style("Target workspace:", fg="bright_blue"),
                 ws_name,
-                f"({current_workspace()})",
+                f"({workspace_id})",
             ])
         )
         click.echo(
@@ -450,7 +462,7 @@ def register(
                 ctx.snakefile,
                 ctx.version,
                 ctx.default_container.image_name,
-                current_workspace(),
+                workspace_id,
             )
         elif ctx.workflow_type == WorkflowType.nextflow:
             assert ctx.nf_script is not None
@@ -571,7 +583,7 @@ def register(
                     ) from e
 
             reg_resp = register_serialized_pkg(
-                protos, ctx.token, ctx.version, current_workspace()
+                protos, ctx.token, ctx.version, workspace_id
             )
             _print_reg_resp(reg_resp, ctx.default_container.image_name)
 
@@ -600,7 +612,7 @@ def register(
                     {
                         "name": wf_name,
                         "version": ctx.version,
-                        "ownerId": current_workspace(),
+                        "ownerId": workspace_id,
                     },
                 )["workflowInfos"]["nodes"]
                 time.sleep(1)