Merge pull request #587 from latchbio/tim/ws

add better ws handling

Author: t-proctor

CHANGELOG.md

@@ -16,6 +16,12 @@ Types of changes
 
 # Latch SDK Changelog
 
+## 2.71.1 - 2026-04-21
+
+### Fixed
+
+* Improve workspace CLI handling when no workspace is selected or available
+
 ## 2.71.0 - 2026-04-02
 
 ### Dependencies

pyproject.toml

@@ -12,7 +12,7 @@ include = ["src/**/*.py", "src/**/py.typed", "src/latch_cli/services/init/*"]
 
 [project]
 name = "latch"
-version = "2.71.0"
+version = "2.71.1"
 description = "The Latch SDK"
 authors = [{ name = "Kenny Workman", email = "kenny@latch.bio" }]
 maintainers = [{ name = "Ayush Kamat", email = "ayush@latch.bio" }]

src/latch/utils.py

@@ -47,6 +47,9 @@ class WSInfo(TypedDict):
     default: bool
 
 
+class NoWorkspaceSelectedError(ValueError): ...
+
+
 def get_workspaces() -> Dict[str, WSInfo]:
     """Retrieve workspaces that user can access.
 
@@ -150,7 +153,7 @@ def current_workspace() -> str:
     """
 
     ws = os.environ.get("LATCH_WORKSPACE")
-    if ws is not None:
+    if ws not in {None, ""}:
         return ws
 
     ws = user_config.workspace_id
@@ -170,13 +173,20 @@ def current_workspace() -> str:
         """),
     )["accountInfoCurrent"]
 
-    ws = res["id"]
+    if res is not None:
+        is_local = os.environ.get("FLYTE_INTERNAL_EXECUTION_ID") is None
+        if is_local and res["user"] is not None:
+            default_account = res["user"]["defaultAccount"]
+            if default_account is not None:
+                return default_account
 
-    is_local = os.environ.get("FLYTE_INTERNAL_EXECUTION_ID") is None
-    if is_local and res["user"] is not None:
-        ws = res["user"]["defaultAccount"]
+        ws = res["id"]
+        if ws is not None:
+            return ws
 
-    return ws
+    raise NoWorkspaceSelectedError(
+        "No workspaces found. Please create a workspace at https://console.latch.bio before proceeding."
+    )
 
 
 class NotFoundError(ValueError): ...

src/latch_cli/main.py

@@ -15,7 +15,7 @@
 
 import latch_cli.click_utils
 from latch.ldata._transfer.progress import Progress as _Progress  # noqa: PLC2701
-from latch.utils import current_workspace
+from latch.utils import NoWorkspaceSelectedError, current_workspace
 from latch_cli.click_utils import EnumChoice
 from latch_cli.exceptions.handler import CrashHandler
 from latch_cli.services.cp.autocomplete import complete as cp_complete
@@ -66,6 +66,21 @@ def decorated(*args: P.args, **kwargs: P.kwargs):
     return decorated
 
 
+def requires_workspace(f: Callable[P, T]) -> Callable[P, T]:
+    def decorated(*args: P.args, **kwargs: P.kwargs):
+        try:
+            current_workspace()
+        except NoWorkspaceSelectedError as e:
+            click.secho(str(e), fg="red")
+            raise click.exceptions.Exit(1) from e
+
+        return f(*args, **kwargs)
+
+    decorated.__doc__ = f.__doc__
+
+    return decorated
+
+
 @click.group("latch", context_settings={"max_content_width": 160})
 @click.version_option(package_name="latch")
 def main():
@@ -485,6 +500,7 @@ def generate_metadata(
     help="Size of machine to provision for develop session",
 )
 @requires_login
+@requires_workspace
 def local_development(
     pkg_root: Path,
     yes: bool,
@@ -533,6 +549,7 @@ def local_development(
     help="Optional container index to inspect (only used for Map Tasks)",
 )
 @requires_login
+@requires_workspace
 def execute(
     execution_id: Optional[str], egn_id: Optional[str], container_index: Optional[int]
 ):
@@ -714,6 +731,7 @@ def image_ls():
     ),
 )
 @requires_login
+@requires_workspace
 def register(
     pkg_root: str,
     disable_auto_version: bool,
@@ -810,6 +828,7 @@ def register(
     help="The version of the workflow to launch. Defaults to latest.",
 )
 @requires_login
+@requires_workspace
 def launch(params_file: Path, version: Union[str, None] = None):
     """[DEPRECATED] Launch a workflow using a python parameter map.
 
@@ -851,6 +870,7 @@ def launch(params_file: Path, version: Union[str, None] = None):
     "--version", default=None, help="The version of the workflow. Defaults to latest."
 )
 @requires_login
+@requires_workspace
 def get_params(wf_name: Union[str, None], version: Union[str, None] = None):
     """[DEPRECATED] Generate a python parameter map for a workflow.
 
@@ -890,6 +910,7 @@ def get_params(wf_name: Union[str, None], version: Union[str, None] = None):
     help="The name of the workflow to list. Will display all versions",
 )
 @requires_login
+@requires_workspace
 def get_wf(name: Union[str, None] = None):
     """List workflows."""
     crash_handler.message = "Unable to get workflows"
@@ -919,6 +940,7 @@ def get_wf(name: Union[str, None] = None):
 @main.command("preview")
 @click.argument("pkg_root", nargs=1, type=click.Path(exists=True, path_type=Path))
 @requires_login
+@requires_workspace
 def preview(pkg_root: Path):
     """Creates a preview of your workflow interface."""
     crash_handler.message = f"Unable to preview inputs for {pkg_root}"
@@ -931,6 +953,7 @@ def preview(pkg_root: Path):
 
 @main.command("get-executions")
 @requires_login
+@requires_workspace
 def get_executions():
     """Spawns an interactive terminal UI that shows all executions in a given workspace"""
 
@@ -1263,6 +1286,7 @@ def generate_entrypoint(
     "--execution-id", "-e", type=str, help="Optional execution ID to inspect."
 )
 @requires_login
+@requires_workspace
 def attach(execution_id: Optional[str]):
     """Drops the user into an interactive shell to inspect the workdir of a nextflow execution."""
 
@@ -1302,6 +1326,7 @@ def attach(execution_id: Optional[str]):
     help="Path to the entrypoint nextflow file. Must be relative to the package root.",
 )
 @requires_login
+@requires_workspace
 def nf_register(
     pkg_root: Path,
     yes: bool,

src/latch_cli/services/workspace.py

@@ -3,7 +3,12 @@
 import click
 from latch_sdk_config.user import user_config
 
-from latch.utils import current_workspace, get_workspaces, WSInfo
+from latch.utils import (
+    NoWorkspaceSelectedError,
+    WSInfo,
+    current_workspace,
+    get_workspaces,
+)
 from latch_cli.menus import SelectOption, select_tui
 
 
@@ -15,7 +20,11 @@ def workspace():
     """
     data = get_workspaces()
 
-    old_id = current_workspace()
+    old_id: str | None
+    try:
+        old_id = current_workspace()
+    except NoWorkspaceSelectedError:
+        old_id = None
 
     selected_marker = "\x1b[3m\x1b[2m (currently selected) \x1b[22m\x1b[23m"
 
@@ -25,7 +34,11 @@ def workspace():
     ):
         options.append(
             {
-                "display_name": info["name"] if old_id != info["workspace_id"] else info["name"] + selected_marker,
+                "display_name": (
+                    info["name"]
+                    if old_id != info["workspace_id"]
+                    else info["name"] + selected_marker
+                ),
                 "value": info,
             }
         )