fix(cli): address code review findings #1, #3, #4

- #1 MUST-FIX: Fix Workspace.create_workflow to map definition->config
  and drop unsupported description param
- #3 SHOULD-FIX: Delegate Project.get_annotation_jobs/get_annotation_job
  to rfapi instead of inline HTTP calls
- #4 SHOULD-FIX: Extract resolve_ws_and_key to _resolver.py, deduplicate
  from folder/workspace/workflow handlers
- Update test to expect exit code 2 (auth error) for missing workspace

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: yeldarby

roboflow/cli/_resolver.py

@@ -113,3 +113,25 @@ def resolve_resource(
         f"Cannot resolve '{shorthand}': expected 1-3 path segments "
         "(project, workspace/project, or workspace/project/version)."
     )
+
+
+def resolve_ws_and_key(args) -> Optional[Tuple[str, str]]:
+    """Resolve workspace and API key from CLI args.
+
+    Returns (workspace_url, api_key) or ``None`` after calling
+    ``output_error`` on failure.
+    """
+    from roboflow.cli._output import output_error
+    from roboflow.config import load_roboflow_api_key
+
+    ws = getattr(args, "workspace", None) or resolve_default_workspace(api_key=getattr(args, "api_key", None))
+    if not ws:
+        output_error(args, "No workspace specified.", hint="Use --workspace or run 'roboflow auth login'.", exit_code=2)
+        return None
+
+    api_key = getattr(args, "api_key", None) or load_roboflow_api_key(ws)
+    if not api_key:
+        output_error(args, "No API key found.", hint="Set ROBOFLOW_API_KEY or run 'roboflow auth login'.", exit_code=2)
+        return None
+
+    return ws, api_key

roboflow/cli/handlers/folder.py

@@ -46,21 +46,9 @@ def register(subparsers: argparse._SubParsersAction) -> None:  # type: ignore[ty
 
 def _resolve_ws_and_key(args: argparse.Namespace):
     """Resolve workspace and API key, returning (ws, api_key) or None on error."""
-    from roboflow.cli._output import output_error
-    from roboflow.cli._resolver import resolve_default_workspace
-    from roboflow.config import load_roboflow_api_key
+    from roboflow.cli._resolver import resolve_ws_and_key
 
-    ws = args.workspace or resolve_default_workspace(api_key=args.api_key)
-    if not ws:
-        output_error(args, "No workspace specified.", hint="Use --workspace or run 'roboflow auth login'.", exit_code=2)
-        return None
-
-    api_key = args.api_key or load_roboflow_api_key(ws)
-    if not api_key:
-        output_error(args, "No API key found.", hint="Set ROBOFLOW_API_KEY or run 'roboflow auth login'.", exit_code=2)
-        return None
-
-    return ws, api_key
+    return resolve_ws_and_key(args)
 
 
 def _list_folders(args: argparse.Namespace) -> None:

roboflow/cli/handlers/workflow.py

@@ -75,24 +75,9 @@ def register(subparsers: argparse._SubParsersAction) -> None:  # type: ignore[ty
 
 def _resolve_workspace_and_key(args: argparse.Namespace):
     """Return (workspace_url, api_key) or call output_error and return None."""
-    from roboflow.cli._output import output_error
-    from roboflow.cli._resolver import resolve_default_workspace
-    from roboflow.config import load_roboflow_api_key
-
-    workspace_url = args.workspace
-    if not workspace_url:
-        workspace_url = resolve_default_workspace(api_key=args.api_key)
-
-    if not workspace_url:
-        output_error(args, "No workspace specified.", hint="Use --workspace or run 'roboflow auth login'.")
-        return None
-
-    api_key = args.api_key or load_roboflow_api_key(workspace_url)
-    if not api_key:
-        output_error(args, "No API key found.", hint="Set ROBOFLOW_API_KEY or run 'roboflow auth login'.", exit_code=2)
-        return None
+    from roboflow.cli._resolver import resolve_ws_and_key
 
-    return workspace_url, api_key
+    return resolve_ws_and_key(args)
 
 
 def _read_definition_file(args: argparse.Namespace):

roboflow/cli/handlers/workspace.py

@@ -133,21 +133,9 @@ def _get_workspace(args: argparse.Namespace) -> None:
 
 def _resolve_ws_and_key(args: argparse.Namespace):
     """Resolve workspace and API key for workspace subcommands."""
-    from roboflow.cli._output import output_error
-    from roboflow.cli._resolver import resolve_default_workspace
-    from roboflow.config import load_roboflow_api_key
-
-    ws = args.workspace or resolve_default_workspace(api_key=args.api_key)
-    if not ws:
-        output_error(args, "No workspace specified.", hint="Use --workspace or run 'roboflow auth login'.", exit_code=2)
-        return None
-
-    api_key = args.api_key or load_roboflow_api_key(ws)
-    if not api_key:
-        output_error(args, "No API key found.", hint="Set ROBOFLOW_API_KEY or run 'roboflow auth login'.", exit_code=2)
-        return None
+    from roboflow.cli._resolver import resolve_ws_and_key
 
-    return ws, api_key
+    return resolve_ws_and_key(args)
 
 
 def _workspace_usage(args: argparse.Namespace) -> None:

roboflow/core/project.py

@@ -878,17 +878,9 @@ def get_annotation_jobs(self) -> Dict:
         Returns:
             Dict: A dictionary containing the list of annotation jobs.
         """
-        url = f"{API_URL}/{self.__workspace}/{self.__project_name}/jobs?api_key={self.__api_key}"
-        response = requests.get(url)
-        if response.status_code != 200:
-            try:
-                error_data = response.json()
-                if "error" in error_data:
-                    raise RuntimeError(error_data["error"])
-                raise RuntimeError(response.text)
-            except ValueError:
-                raise RuntimeError(f"Failed to get annotation jobs: {response.text}")
-        return response.json()
+        from roboflow.adapters import rfapi
+
+        return rfapi.list_annotation_jobs(self.__api_key, self.__workspace, self.__project_name)
 
     def get_annotation_job(self, job_id: str) -> Dict:
         """Get information for a specific annotation job.
@@ -899,17 +891,9 @@ def get_annotation_job(self, job_id: str) -> Dict:
         Returns:
             Dict: A dictionary containing the job details.
         """
-        url = f"{API_URL}/{self.__workspace}/{self.__project_name}/jobs/{job_id}?api_key={self.__api_key}"
-        response = requests.get(url)
-        if response.status_code != 200:
-            try:
-                error_data = response.json()
-                if "error" in error_data:
-                    raise RuntimeError(error_data["error"])
-                raise RuntimeError(response.text)
-            except ValueError:
-                raise RuntimeError(f"Failed to get annotation job: {response.text}")
-        return response.json()
+        from roboflow.adapters import rfapi
+
+        return rfapi.get_annotation_job(self.__api_key, self.__workspace, self.__project_name, job_id)
 
     def create_annotation_job(
         self, name: str, batch_id: str, num_images: int, labeler_email: str, reviewer_email: str

roboflow/core/workspace.py

@@ -913,13 +913,14 @@ def get_workflow(self, workflow_url):
 
         return rfapi.get_workflow(self.__api_key, self.url, workflow_url)
 
-    def create_workflow(self, name, definition=None, description=None):
+    def create_workflow(self, name, definition=None):
         """Create a new workflow."""
+        import json
+
         from roboflow.adapters import rfapi
 
-        return rfapi.create_workflow(
-            self.__api_key, self.url, name=name, definition=definition, description=description
-        )
+        config = json.dumps(definition) if definition else None
+        return rfapi.create_workflow(self.__api_key, self.url, name=name, config=config)
 
     # -----------------------------------------------------------------
     # Phase 2: Workspace statistics

tests/cli/test_workflow_handler.py

@@ -408,7 +408,7 @@ def test_list_no_workspace(self, _mock_resolve):
         args = _make_args(workspace=None, api_key=None)
         with self.assertRaises(SystemExit) as ctx:
             _list_workflows(args)
-        self.assertEqual(ctx.exception.code, 1)
+        self.assertEqual(ctx.exception.code, 2)
 
 
 if __name__ == "__main__":