From f82fdbb16a9894c4adba0f22c72ecbf6738f8da2 Mon Sep 17 00:00:00 2001 From: Leo Ueno Date: Mon, 24 Aug 2026 01:43:57 -0700 Subject: [PATCH 1/4] Add Batch Processing CLI commands --- CLI-COMMANDS.md | 21 ++ roboflow/adapters/rfapi.py | 119 +++++++ roboflow/cli/__init__.py | 2 +- roboflow/cli/handlers/batch.py | 300 ++++++++++++++++-- tests/adapters/test_rfapi_batch_processing.py | 56 ++++ tests/cli/test_batch_handler.py | 187 ++++++++++- tests/cli/test_completion_handler.py | 1 - 7 files changed, 640 insertions(+), 46 deletions(-) create mode 100644 tests/adapters/test_rfapi_batch_processing.py diff --git a/CLI-COMMANDS.md b/CLI-COMMANDS.md index 763cb790..960a4489 100644 --- a/CLI-COMMANDS.md +++ b/CLI-COMMANDS.md @@ -47,6 +47,27 @@ roboflow download my-workspace/my-project/3 -f coco # alias roboflow infer photo.jpg -m my-project/3 ``` +### Batch Process Asset Library images + +```bash +# Exact reviewed selection (CPU is the product default): +roboflow batch create --workflow inspect-defects --image-ids img_1,img_2 + +# Or every current match for a structured RoboQL filter: +roboflow batch create --workflow inspect-defects --query "tag:night-shift" + +# Monitor and control the durable job: +roboflow batch status +roboflow batch list +roboflow batch abort +roboflow batch restart +``` + +The create response includes `taskId`, `jobId`, and `requestId`. A job continues if the terminal +closes. If a create request has an ambiguous network result, retry with the same `--request-id` to +avoid a duplicate. Local folders must first be uploaded into Roboflow; Batch Processing never sends +local file contents through an Agent or CLI job-configuration request. + ### Train, monitor, cancel, stop ```bash diff --git a/roboflow/adapters/rfapi.py b/roboflow/adapters/rfapi.py index 16d29a66..ee368358 100644 --- a/roboflow/adapters/rfapi.py +++ b/roboflow/adapters/rfapi.py @@ -1710,6 +1710,125 @@ def get_video_job_status(api_key, job_id): return response.json() +# --------------------------------------------------------------------------- +# Batch Processing (Asset Library orchestration) +# --------------------------------------------------------------------------- + + +def _batch_processing_url(workspace_url, suffix=""): + return f"{API_URL}/batch-processing/v1/external/{workspace_url}/asset-library/jobs{suffix}" + + +def _batch_processing_headers(api_key): + # Keep credentials out of URLs, proxy logs, and shell history. validateToken supports Bearer. + return {"Authorization": f"Bearer {api_key}"} + + +def _raise_for_batch_processing_response(response): + message = response.text + try: + body = response.json() + if isinstance(body, dict): + error = body.get("error") + if isinstance(error, dict): + message = error.get("message") or error.get("hint") or message + elif error: + message = str(error) + else: + message = body.get("message") or message + except (TypeError, ValueError): + pass + raise RoboflowError(message, status_code=response.status_code) + + +def create_asset_library_batch_job( + api_key, + workspace_url, + *, + workflow_id, + idempotency_key, + image_ids=None, + query=None, + machine_type="cpu", + display_name=None, +): + """Queue a published Workflow over an exact Asset Library selection.""" + payload = { + "workflowId": workflow_id, + "idempotencyKey": idempotency_key, + "machineType": machine_type, + } + if image_ids is not None: + payload["imageIds"] = image_ids + if query is not None: + payload["query"] = query + if display_name: + payload["displayName"] = display_name + response = requests.post( + _batch_processing_url(workspace_url), + headers=_batch_processing_headers(api_key), + json=payload, + ) + if response.status_code != 202: + _raise_for_batch_processing_response(response) + return response.json() + + +def list_batch_processing_jobs(api_key, workspace_url, *, page_size=10, next_page_token=None, search=None): + """List durable Batch Processing jobs in a workspace.""" + params = {"pageSize": page_size} + if next_page_token: + params["nextPageToken"] = next_page_token + if search: + params["search"] = search + response = requests.get( + _batch_processing_url(workspace_url), + headers=_batch_processing_headers(api_key), + params=params, + ) + if response.status_code != 200: + _raise_for_batch_processing_response(response) + return response.json() + + +def get_batch_processing_job(api_key, workspace_url, job_id): + """Get current metadata for one Batch Processing job.""" + encoded = quote(job_id, safe="") + response = requests.get( + _batch_processing_url(workspace_url, f"/{encoded}"), + headers=_batch_processing_headers(api_key), + ) + if response.status_code != 200: + _raise_for_batch_processing_response(response) + return response.json() + + +def abort_batch_processing_job(api_key, workspace_url, job_id): + """Abort one Batch Processing job.""" + encoded = quote(job_id, safe="") + response = requests.post( + _batch_processing_url(workspace_url, f"/{encoded}/abort"), + headers=_batch_processing_headers(api_key), + json={}, + ) + if response.status_code != 200: + _raise_for_batch_processing_response(response) + return response.json() + + +def restart_batch_processing_job(api_key, workspace_url, job_id): + """Restart one Batch Processing job with its existing configuration.""" + encoded = quote(job_id, safe="") + response = requests.post( + _batch_processing_url(workspace_url, f"/{encoded}/restart"), + headers=_batch_processing_headers(api_key), + json={}, + ) + if response.status_code != 200: + _raise_for_batch_processing_response(response) + return response.json() + + # --------------------------------------------------------------------------- # Phase 2: Universe search # --------------------------------------------------------------------------- diff --git a/roboflow/cli/__init__.py b/roboflow/cli/__init__.py index 54754a08..3befa965 100644 --- a/roboflow/cli/__init__.py +++ b/roboflow/cli/__init__.py @@ -213,7 +213,7 @@ def _walk(group: Any, prefix: str = "") -> None: app.add_typer(api_key_app, name="api-key") app.add_typer(asynctasks_app, name="asynctasks") app.add_typer(auth_app, name="auth") -app.add_typer(batch_app, name="batch", hidden=True) # All stubs — hidden until implemented +app.add_typer(batch_app, name="batch") app.add_typer(completion_app, name="completion") app.add_typer(deployment_app, name="deployment") app.add_typer(device_app, name="device") diff --git a/roboflow/cli/handlers/batch.py b/roboflow/cli/handlers/batch.py index 31d24647..dfd29940 100644 --- a/roboflow/cli/handlers/batch.py +++ b/roboflow/cli/handlers/batch.py @@ -1,33 +1,64 @@ -"""Batch processing commands.""" +"""Batch Processing commands backed by Roboflow's durable workspace jobs.""" from __future__ import annotations +import re +import uuid +from enum import Enum from typing import Annotated, Optional import typer from roboflow.cli._compat import SortedGroup, ctx_to_args -batch_app = typer.Typer(cls=SortedGroup, help="Batch processing operations", no_args_is_help=True) +batch_app = typer.Typer(cls=SortedGroup, help="Run and manage Batch Processing jobs", no_args_is_help=True) -def _stub(args) -> None: # noqa: ANN001 - from roboflow.cli._output import output_error +class BatchMachine(str, Enum): + """Execution pools exposed by the Asset Library Batch Processing surface.""" - output_error(args, "This command is not yet implemented.", hint="Coming soon.", exit_code=1) + CPU = "cpu" + GPU = "gpu" @batch_app.command("create") def create( ctx: typer.Context, - workflow: Annotated[str, typer.Option(help="Workflow ID to run")], - input: Annotated[str, typer.Option(help="Input path (image directory or video file)")], - model: Annotated[Optional[str], typer.Option(help="Model ID override (default: workflow model)")] = None, - output_dir: Annotated[Optional[str], typer.Option("--output", help="Output directory for results")] = None, + workflow: Annotated[str, typer.Option(help="Published Workflow ID to run")], + image_ids: Annotated[ + Optional[str], + typer.Option("--image-ids", help="Comma-separated exact Asset Library image IDs"), + ] = None, + query: Annotated[ + Optional[str], + typer.Option(help="Reviewed structured RoboQL filter selecting all current matches"), + ] = None, + all_images: Annotated[ + bool, + typer.Option("--all", help="Explicitly run on the entire Asset Library"), + ] = False, + machine: Annotated[ + BatchMachine, + typer.Option(help="Execution machine; defaults to the product UI default"), + ] = BatchMachine.CPU, + name: Annotated[Optional[str], typer.Option(help="Optional user-facing job name")] = None, + request_id: Annotated[ + Optional[str], + typer.Option(help="Stable idempotency key; reuse only when retrying this same launch"), + ] = None, ) -> None: - """Create a batch processing job.""" - args = ctx_to_args(ctx, workflow=workflow, input=input, model=model, output=output_dir) - _stub(args) + """Queue a Workflow over one exact or reviewed Asset Library selection.""" + args = ctx_to_args( + ctx, + workflow=workflow, + image_ids=image_ids, + query=query, + all_images=all_images, + machine=machine.value, + name=name, + request_id=request_id, + ) + _create(args) @batch_app.command("status") @@ -35,29 +66,244 @@ def status( ctx: typer.Context, job_id: Annotated[str, typer.Argument(help="Batch job ID")], ) -> None: - """Check batch job status.""" - args = ctx_to_args(ctx, job_id=job_id) - _stub(args) + """Show current durable job status and configuration.""" + _status(ctx_to_args(ctx, job_id=job_id)) @batch_app.command("list") def list_jobs( ctx: typer.Context, - status_filter: Annotated[ - Optional[str], typer.Option("--status", help="Filter by status (pending, running, completed, failed)") - ] = None, + page_size: Annotated[int, typer.Option(min=1, max=100, help="Jobs per page")] = 10, + next_page_token: Annotated[Optional[str], typer.Option(help="Pagination token")] = None, + search: Annotated[Optional[str], typer.Option(help="Search names, Workflows, and status text")] = None, ) -> None: - """List batch jobs.""" - args = ctx_to_args(ctx, status=status_filter) - _stub(args) + """List Batch Processing jobs.""" + _list(ctx_to_args(ctx, page_size=page_size, next_page_token=next_page_token, search=search)) -@batch_app.command("results") -def results( +@batch_app.command("abort") +def abort( ctx: typer.Context, job_id: Annotated[str, typer.Argument(help="Batch job ID")], - format: Annotated[Optional[str], typer.Option(help="Output format (json, csv)")] = None, + yes: Annotated[bool, typer.Option("--yes", "-y", help="Confirm without prompting")] = False, ) -> None: - """Get batch job results.""" - args = ctx_to_args(ctx, job_id=job_id, format=format) - _stub(args) + """Abort a Batch Processing job.""" + _abort(ctx_to_args(ctx, job_id=job_id, yes=yes)) + + +@batch_app.command("restart") +def restart( + ctx: typer.Context, + job_id: Annotated[str, typer.Argument(help="Batch job ID")], + yes: Annotated[bool, typer.Option("--yes", "-y", help="Confirm credit-spending restart")] = False, +) -> None: + """Restart a Batch Processing job with its existing configuration.""" + _restart(ctx_to_args(ctx, job_id=job_id, yes=yes)) + + +def _resolve_ws_and_key(args): # noqa: ANN001 + from roboflow.cli._resolver import resolve_ws_and_key + + return resolve_ws_and_key(args) + + +def _parse_image_ids(raw: Optional[str]) -> list[str]: + if raw is None: + return [] + return list(dict.fromkeys(part.strip() for part in raw.split(",") if part.strip())) + + +def _validate_job_id(args, job_id: str, *, label: str = "job ID") -> None: # noqa: ANN001 + from roboflow.cli._output import output_error + + if not re.fullmatch(r"[a-z0-9-]{1,20}", job_id): + output_error( + args, + f"{label} must be 1-20 lowercase letters, numbers, or hyphens.", + ) + + +def _create(args) -> None: # noqa: ANN001 + from roboflow.adapters import rfapi + from roboflow.cli._output import output, output_api_error, output_error + + ids = _parse_image_ids(args.image_ids) + selection_modes = int(bool(ids)) + int(args.query is not None) + int(args.all_images) + if selection_modes != 1: + output_error( + args, + "Choose exactly one selection: --image-ids, --query, or --all.", + hint="The CLI never broadens an omitted or ambiguous selection to the whole Asset Library.", + ) + return + if args.query is not None and not args.query.strip(): + output_error( + args, + "--query cannot be empty.", + hint="Use --all to explicitly select the entire Asset Library.", + ) + return + if len(ids) > 2048: + output_error(args, "At most 2048 explicit image IDs can be queued in one request.") + return + + request_id = args.request_id or str(uuid.uuid4()) + if not 8 <= len(request_id) <= 128 or not re.fullmatch(r"[A-Za-z0-9_-]+", request_id): + output_error( + args, + "--request-id must be 8-128 letters, numbers, underscores, or hyphens.", + ) + return + resolved = _resolve_ws_and_key(args) + if not resolved: + return + workspace, api_key = resolved + query = "" if args.all_images else args.query + + try: + result = rfapi.create_asset_library_batch_job( + api_key, + workspace, + workflow_id=args.workflow, + idempotency_key=request_id, + image_ids=ids or None, + query=query, + machine_type=args.machine, + display_name=args.name, + ) + except rfapi.RoboflowError as exc: + output_api_error( + args, + exc, + hint=f"Retry the same launch with --request-id {request_id}; use a new ID only for a new job intent.", + auth_hint="Check the API key has 'batch-processing:trigger' scope and access to the selected resources.", + ) + return + + result = {**result, "requestId": request_id} + text = ( + f"Queued {result.get('displayName') or result.get('jobId')}\n" + f"jobId={result.get('jobId')}\n" + f"taskId={result.get('taskId')}\n" + f"requestId={request_id}\n" + f"Next: roboflow asynctasks wait {result.get('taskId')}" + ) + output(args, result, text=text) + + +def _status(args) -> None: # noqa: ANN001 + from roboflow.adapters import rfapi + from roboflow.cli._output import output, output_api_error + + _validate_job_id(args, args.job_id) + resolved = _resolve_ws_and_key(args) + if not resolved: + return + workspace, api_key = resolved + try: + result = rfapi.get_batch_processing_job(api_key, workspace, args.job_id) + except rfapi.RoboflowError as exc: + output_api_error( + args, + exc, + auth_hint="Check the API key has 'batch-processing:read' scope.", + not_found_hint="Check the job ID and workspace.", + ) + return + + job = result.get("job", {}) + state = job.get("currentStage") or ("terminal" if job.get("isTerminal") else "queued") + output( + args, + result, + text=( + f"jobId={job.get('jobId', args.job_id)} state={state} " + f"terminal={job.get('isTerminal')} error={job.get('error')}" + ), + ) + + +def _list(args) -> None: # noqa: ANN001 + from roboflow.adapters import rfapi + from roboflow.cli._output import output, output_api_error, output_error + from roboflow.cli._table import format_table + + if args.next_page_token: + _validate_job_id(args, args.next_page_token, label="--next-page-token") + if args.search is not None and len(args.search) > 160: + output_error(args, "--search must be at most 160 characters.") + return + resolved = _resolve_ws_and_key(args) + if not resolved: + return + workspace, api_key = resolved + try: + result = rfapi.list_batch_processing_jobs( + api_key, + workspace, + page_size=args.page_size, + next_page_token=args.next_page_token, + search=args.search, + ) + except rfapi.RoboflowError as exc: + output_api_error(args, exc, auth_hint="Check the API key has 'batch-processing:read' scope.") + return + + rows = [ + { + "jobId": job.get("jobId", ""), + "name": job.get("name", ""), + "stage": job.get("currentStage") or ("terminal" if job.get("isTerminal") else "queued"), + "error": job.get("error", False), + "updated": job.get("lastUpdate", ""), + } + for job in result.get("jobs", []) + ] + table = format_table(rows, columns=["jobId", "name", "stage", "error", "updated"]) + if result.get("nextPageToken"): + table += f"\nNext page: --next-page-token {result['nextPageToken']}" + output(args, result, text=table) + + +def _abort(args) -> None: # noqa: ANN001 + from roboflow.adapters import rfapi + from roboflow.cli._output import confirm_destructive + + _validate_job_id(args, args.job_id) + if not confirm_destructive(args, f"Abort Batch Processing job '{args.job_id}'?"): + return + _run_control_action(args, rfapi.abort_batch_processing_job, "Aborted") + + +def _restart(args) -> None: # noqa: ANN001 + from roboflow.adapters import rfapi + from roboflow.cli._output import confirm_destructive + + _validate_job_id(args, args.job_id) + if not confirm_destructive( + args, + f"Restart Batch Processing job '{args.job_id}'? This can consume credits.", + ): + return + _run_control_action(args, rfapi.restart_batch_processing_job, "Restarted") + + +def _run_control_action(args, action, verb: str) -> None: # noqa: ANN001 + from roboflow.adapters import rfapi + from roboflow.cli._output import output, output_api_error + + resolved = _resolve_ws_and_key(args) + if not resolved: + return + workspace, api_key = resolved + try: + result = action(api_key, workspace, args.job_id) + except rfapi.RoboflowError as exc: + output_api_error( + args, + exc, + auth_hint="Check the API key has 'batch-processing:trigger' scope.", + not_found_hint="Check the job ID and workspace.", + ) + return + output(args, result, text=f"{verb} Batch Processing job {args.job_id}.") diff --git a/tests/adapters/test_rfapi_batch_processing.py b/tests/adapters/test_rfapi_batch_processing.py new file mode 100644 index 00000000..ce65a2e4 --- /dev/null +++ b/tests/adapters/test_rfapi_batch_processing.py @@ -0,0 +1,56 @@ +"""HTTP contract tests for Asset Library Batch Processing adapters.""" + +from __future__ import annotations + +import unittest +from unittest.mock import Mock, patch + +from roboflow.adapters import rfapi + + +class TestBatchProcessingAdapter(unittest.TestCase): + @patch("roboflow.adapters.rfapi.requests.post") + def test_create_uses_bearer_auth_and_idempotent_payload(self, mock_post) -> None: + response = Mock(status_code=202) + response.json.return_value = {"status": "queued", "jobId": "al-123"} + mock_post.return_value = response + + result = rfapi.create_asset_library_batch_job( + "private-key", + "workspace-1", + workflow_id="workflow-1", + idempotency_key="request-123", + image_ids=["image-1"], + ) + + self.assertEqual(result["jobId"], "al-123") + _, kwargs = mock_post.call_args + self.assertEqual(kwargs["headers"], {"Authorization": "Bearer private-key"}) + self.assertNotIn("private-key", mock_post.call_args.args[0]) + self.assertEqual(kwargs["json"]["idempotencyKey"], "request-123") + + @patch("roboflow.adapters.rfapi.requests.get") + def test_status_encodes_untrusted_job_id(self, mock_get) -> None: + response = Mock(status_code=200) + response.json.return_value = {"status": "ok", "job": {"jobId": "bad/id"}} + mock_get.return_value = response + + rfapi.get_batch_processing_job("private-key", "workspace-1", "bad/id") + + self.assertTrue(mock_get.call_args.args[0].endswith("/bad%2Fid")) + + @patch("roboflow.adapters.rfapi.requests.get") + def test_error_preserves_http_status_for_cli_exit_codes(self, mock_get) -> None: + response = Mock(status_code=404, text='{"error":{"message":"Job not found"}}') + response.json.return_value = {"error": {"message": "Job not found"}} + mock_get.return_value = response + + with self.assertRaises(rfapi.RoboflowError) as ctx: + rfapi.get_batch_processing_job("private-key", "workspace-1", "missing") + + self.assertEqual(ctx.exception.status_code, 404) + self.assertEqual(str(ctx.exception), "Job not found") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/cli/test_batch_handler.py b/tests/cli/test_batch_handler.py index bfe773d1..bc605489 100644 --- a/tests/cli/test_batch_handler.py +++ b/tests/cli/test_batch_handler.py @@ -1,37 +1,190 @@ -"""Tests for the batch CLI handler.""" +"""Tests for the durable Batch Processing CLI.""" +from __future__ import annotations + +import json import unittest +from unittest.mock import patch from typer.testing import CliRunner from roboflow.cli import app runner = CliRunner() +BASE = ["--workspace", "workspace-1", "--api-key", "private-key"] class TestBatchRegistration(unittest.TestCase): - """Verify batch handler registers expected subcommands.""" + """Batch commands are public and documented by Typer.""" + + def test_batch_is_visible_in_root_help(self) -> None: + result = runner.invoke(app, ["--help"]) + self.assertEqual(result.exit_code, 0, result.output) + self.assertIn("batch", result.output) + self.assertIn("Run and manage Batch Processing jobs", result.output) + + def test_batch_subcommands(self) -> None: + for verb in ("create", "status", "list", "abort", "restart"): + with self.subTest(verb=verb): + result = runner.invoke(app, ["batch", verb, "--help"]) + self.assertEqual(result.exit_code, 0, result.output) + + +class TestBatchCreate(unittest.TestCase): + @patch("roboflow.adapters.rfapi.create_asset_library_batch_job") + def test_create_exact_selection_has_stable_machine_output(self, mock_create) -> None: + mock_create.return_value = { + "status": "queued", + "taskId": "task-1", + "jobId": "al-123", + "batchId": "asset-library-123", + "displayName": "Night defects", + } + + result = runner.invoke( + app, + [ + *BASE, + "--json", + "batch", + "create", + "--workflow", + "workflow-1", + "--image-ids", + "image-1,image-1,image-2", + "--request-id", + "request-123", + ], + ) + + self.assertEqual(result.exit_code, 0, result.output) + payload = json.loads(result.output) + self.assertEqual(payload["requestId"], "request-123") + mock_create.assert_called_once_with( + "private-key", + "workspace-1", + workflow_id="workflow-1", + idempotency_key="request-123", + image_ids=["image-1", "image-2"], + query=None, + machine_type="cpu", + display_name=None, + ) + + @patch("roboflow.adapters.rfapi.create_asset_library_batch_job") + def test_all_is_explicit_empty_query_not_an_omitted_selection(self, mock_create) -> None: + mock_create.return_value = {"taskId": "task-1", "jobId": "al-123"} + + result = runner.invoke( + app, + [*BASE, "batch", "create", "--workflow", "workflow-1", "--all"], + ) + + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual(mock_create.call_args.kwargs["query"], "") + self.assertIsNone(mock_create.call_args.kwargs["image_ids"]) + + @patch("roboflow.adapters.rfapi.create_asset_library_batch_job") + def test_ambiguous_selection_fails_closed(self, mock_create) -> None: + result = runner.invoke( + app, + [*BASE, "batch", "create", "--workflow", "workflow-1"], + ) + + self.assertEqual(result.exit_code, 1) + self.assertIn("exactly one selection", result.output) + mock_create.assert_not_called() + + @patch("roboflow.adapters.rfapi.create_asset_library_batch_job") + def test_empty_query_does_not_implicitly_select_every_image(self, mock_create) -> None: + result = runner.invoke( + app, + [*BASE, "batch", "create", "--workflow", "workflow-1", "--query", ""], + ) + + self.assertEqual(result.exit_code, 1) + self.assertIn("--query cannot be empty", result.output) + self.assertIn("Use --all", result.output) + mock_create.assert_not_called() + + @patch("roboflow.adapters.rfapi.create_asset_library_batch_job") + def test_unsafe_request_id_fails_before_network(self, mock_create) -> None: + result = runner.invoke( + app, + [ + *BASE, + "batch", + "create", + "--workflow", + "workflow-1", + "--all", + "--request-id", + "unsafe/key", + ], + ) + + self.assertEqual(result.exit_code, 1) + self.assertIn("--request-id must be", result.output) + mock_create.assert_not_called() + + +class TestBatchLifecycle(unittest.TestCase): + @patch("roboflow.adapters.rfapi.get_batch_processing_job") + def test_status_rejects_malformed_job_id_before_network(self, mock_get) -> None: + result = runner.invoke(app, [*BASE, "batch", "status", "unsafe/job"]) + + self.assertEqual(result.exit_code, 1) + self.assertIn("job ID must be", result.output) + mock_get.assert_not_called() + + @patch("roboflow.adapters.rfapi.get_batch_processing_job") + def test_status_json_is_api_faithful(self, mock_get) -> None: + api_result = { + "status": "ok", + "job": {"jobId": "al-123", "currentStage": "inference", "isTerminal": False, "error": False}, + } + mock_get.return_value = api_result + + result = runner.invoke(app, [*BASE, "--json", "batch", "status", "al-123"]) + + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual(json.loads(result.output), api_result) + + @patch("roboflow.adapters.rfapi.list_batch_processing_jobs") + def test_list_passes_pagination_and_search(self, mock_list) -> None: + mock_list.return_value = {"status": "ok", "jobs": [], "nextPageToken": None} + + result = runner.invoke( + app, + [*BASE, "batch", "list", "--page-size", "25", "--next-page-token", "next-1", "--search", "night"], + ) + + self.assertEqual(result.exit_code, 0, result.output) + mock_list.assert_called_once_with( + "private-key", + "workspace-1", + page_size=25, + next_page_token="next-1", + search="night", + ) - def test_batch_app_exists(self) -> None: - from roboflow.cli.handlers.batch import batch_app + @patch("roboflow.adapters.rfapi.abort_batch_processing_job") + def test_abort_requires_and_honors_explicit_confirmation(self, mock_abort) -> None: + mock_abort.return_value = {"status": "ok", "jobId": "al-123"} - self.assertIsNotNone(batch_app) + result = runner.invoke(app, [*BASE, "batch", "abort", "al-123", "--yes"]) - def test_batch_create_exists(self) -> None: - result = runner.invoke(app, ["batch", "create", "--help"]) - self.assertEqual(result.exit_code, 0) + self.assertEqual(result.exit_code, 0, result.output) + mock_abort.assert_called_once_with("private-key", "workspace-1", "al-123") - def test_batch_status_exists(self) -> None: - result = runner.invoke(app, ["batch", "status", "--help"]) - self.assertEqual(result.exit_code, 0) + @patch("roboflow.adapters.rfapi.restart_batch_processing_job") + def test_restart_requires_and_honors_credit_confirmation(self, mock_restart) -> None: + mock_restart.return_value = {"status": "ok", "jobId": "al-123"} - def test_batch_list_exists(self) -> None: - result = runner.invoke(app, ["batch", "list", "--help"]) - self.assertEqual(result.exit_code, 0) + result = runner.invoke(app, [*BASE, "batch", "restart", "al-123", "--yes"]) - def test_batch_results_exists(self) -> None: - result = runner.invoke(app, ["batch", "results", "--help"]) - self.assertEqual(result.exit_code, 0) + self.assertEqual(result.exit_code, 0, result.output) + mock_restart.assert_called_once_with("private-key", "workspace-1", "al-123") if __name__ == "__main__": diff --git a/tests/cli/test_completion_handler.py b/tests/cli/test_completion_handler.py index 41a57ed6..d701543e 100644 --- a/tests/cli/test_completion_handler.py +++ b/tests/cli/test_completion_handler.py @@ -95,7 +95,6 @@ def test_hidden_commands_filtered_from_completion(self) -> None: "get_workspace_info", "run_video_inference_api", "help", - "batch", } leaked = hidden_examples & visible self.assertFalse(leaked, f"Hidden commands leaked into completion: {leaked}") From b8a0752e46cbfae2035725985a9da4a3c46387e3 Mon Sep 17 00:00:00 2001 From: Leo Ueno Date: Tue, 1 Sep 2026 02:27:20 -0700 Subject: [PATCH 2/4] Use canonical Batch Processing lifecycle routes --- roboflow/adapters/rfapi.py | 18 +++++++++------ tests/adapters/test_rfapi_batch_processing.py | 22 ++++++++++++++++++- 2 files changed, 32 insertions(+), 8 deletions(-) diff --git a/roboflow/adapters/rfapi.py b/roboflow/adapters/rfapi.py index ee368358..eeda35a7 100644 --- a/roboflow/adapters/rfapi.py +++ b/roboflow/adapters/rfapi.py @@ -1715,8 +1715,12 @@ def get_video_job_status(api_key, job_id): # --------------------------------------------------------------------------- -def _batch_processing_url(workspace_url, suffix=""): - return f"{API_URL}/batch-processing/v1/external/{workspace_url}/asset-library/jobs{suffix}" +def _batch_processing_jobs_url(workspace_url, suffix=""): + return f"{API_URL}/batch-processing/v1/external/{workspace_url}/jobs{suffix}" + + +def _asset_library_batch_processing_url(workspace_url): + return f"{API_URL}/batch-processing/v1/external/{workspace_url}/asset-library/jobs" def _batch_processing_headers(api_key): @@ -1765,7 +1769,7 @@ def create_asset_library_batch_job( if display_name: payload["displayName"] = display_name response = requests.post( - _batch_processing_url(workspace_url), + _asset_library_batch_processing_url(workspace_url), headers=_batch_processing_headers(api_key), json=payload, ) @@ -1782,7 +1786,7 @@ def list_batch_processing_jobs(api_key, workspace_url, *, page_size=10, next_pag if search: params["search"] = search response = requests.get( - _batch_processing_url(workspace_url), + _batch_processing_jobs_url(workspace_url), headers=_batch_processing_headers(api_key), params=params, ) @@ -1795,7 +1799,7 @@ def get_batch_processing_job(api_key, workspace_url, job_id): """Get current metadata for one Batch Processing job.""" encoded = quote(job_id, safe="") response = requests.get( - _batch_processing_url(workspace_url, f"/{encoded}"), + _batch_processing_jobs_url(workspace_url, f"/{encoded}"), headers=_batch_processing_headers(api_key), ) if response.status_code != 200: @@ -1807,7 +1811,7 @@ def abort_batch_processing_job(api_key, workspace_url, job_id): """Abort one Batch Processing job.""" encoded = quote(job_id, safe="") response = requests.post( - _batch_processing_url(workspace_url, f"/{encoded}/abort"), + _batch_processing_jobs_url(workspace_url, f"/{encoded}/abort"), headers=_batch_processing_headers(api_key), json={}, ) @@ -1820,7 +1824,7 @@ def restart_batch_processing_job(api_key, workspace_url, job_id): """Restart one Batch Processing job with its existing configuration.""" encoded = quote(job_id, safe="") response = requests.post( - _batch_processing_url(workspace_url, f"/{encoded}/restart"), + _batch_processing_jobs_url(workspace_url, f"/{encoded}/restart"), headers=_batch_processing_headers(api_key), json={}, ) diff --git a/tests/adapters/test_rfapi_batch_processing.py b/tests/adapters/test_rfapi_batch_processing.py index ce65a2e4..9415fd6c 100644 --- a/tests/adapters/test_rfapi_batch_processing.py +++ b/tests/adapters/test_rfapi_batch_processing.py @@ -27,8 +27,25 @@ def test_create_uses_bearer_auth_and_idempotent_payload(self, mock_post) -> None _, kwargs = mock_post.call_args self.assertEqual(kwargs["headers"], {"Authorization": "Bearer private-key"}) self.assertNotIn("private-key", mock_post.call_args.args[0]) + self.assertEqual( + mock_post.call_args.args[0], + f"{rfapi.API_URL}/batch-processing/v1/external/workspace-1/asset-library/jobs", + ) self.assertEqual(kwargs["json"]["idempotencyKey"], "request-123") + @patch("roboflow.adapters.rfapi.requests.get") + def test_list_uses_canonical_jobs_endpoint(self, mock_get) -> None: + response = Mock(status_code=200) + response.json.return_value = {"status": "ok", "jobs": []} + mock_get.return_value = response + + rfapi.list_batch_processing_jobs("private-key", "workspace-1") + + self.assertEqual( + mock_get.call_args.args[0], + f"{rfapi.API_URL}/batch-processing/v1/external/workspace-1/jobs", + ) + @patch("roboflow.adapters.rfapi.requests.get") def test_status_encodes_untrusted_job_id(self, mock_get) -> None: response = Mock(status_code=200) @@ -37,7 +54,10 @@ def test_status_encodes_untrusted_job_id(self, mock_get) -> None: rfapi.get_batch_processing_job("private-key", "workspace-1", "bad/id") - self.assertTrue(mock_get.call_args.args[0].endswith("/bad%2Fid")) + self.assertEqual( + mock_get.call_args.args[0], + f"{rfapi.API_URL}/batch-processing/v1/external/workspace-1/jobs/bad%2Fid", + ) @patch("roboflow.adapters.rfapi.requests.get") def test_error_preserves_http_status_for_cli_exit_codes(self, mock_get) -> None: From c74e69e778c37ce7a9e0d1754beee352fcea32e0 Mon Sep 17 00:00:00 2001 From: Leo Ueno Date: Tue, 1 Sep 2026 16:48:40 -0700 Subject: [PATCH 3/4] Preserve batch request IDs on transport failures --- roboflow/adapters/rfapi.py | 23 +++++++++++++++---- tests/adapters/test_rfapi_batch_processing.py | 21 +++++++++++++++++ tests/cli/test_batch_handler.py | 18 +++++++++++++++ 3 files changed, 57 insertions(+), 5 deletions(-) diff --git a/roboflow/adapters/rfapi.py b/roboflow/adapters/rfapi.py index eeda35a7..1b213725 100644 --- a/roboflow/adapters/rfapi.py +++ b/roboflow/adapters/rfapi.py @@ -1728,6 +1728,14 @@ def _batch_processing_headers(api_key): return {"Authorization": f"Bearer {api_key}"} +def _batch_processing_request(request, *args, **kwargs): + """Make a Batch Processing request with CLI-safe transport errors.""" + try: + return request(*args, **kwargs) + except RequestException as exc: + raise RoboflowError(str(exc)) from exc + + def _raise_for_batch_processing_response(response): message = response.text try: @@ -1768,7 +1776,8 @@ def create_asset_library_batch_job( payload["query"] = query if display_name: payload["displayName"] = display_name - response = requests.post( + response = _batch_processing_request( + requests.post, _asset_library_batch_processing_url(workspace_url), headers=_batch_processing_headers(api_key), json=payload, @@ -1785,7 +1794,8 @@ def list_batch_processing_jobs(api_key, workspace_url, *, page_size=10, next_pag params["nextPageToken"] = next_page_token if search: params["search"] = search - response = requests.get( + response = _batch_processing_request( + requests.get, _batch_processing_jobs_url(workspace_url), headers=_batch_processing_headers(api_key), params=params, @@ -1798,7 +1808,8 @@ def list_batch_processing_jobs(api_key, workspace_url, *, page_size=10, next_pag def get_batch_processing_job(api_key, workspace_url, job_id): """Get current metadata for one Batch Processing job.""" encoded = quote(job_id, safe="") - response = requests.get( + response = _batch_processing_request( + requests.get, _batch_processing_jobs_url(workspace_url, f"/{encoded}"), headers=_batch_processing_headers(api_key), ) @@ -1810,7 +1821,8 @@ def get_batch_processing_job(api_key, workspace_url, job_id): def abort_batch_processing_job(api_key, workspace_url, job_id): """Abort one Batch Processing job.""" encoded = quote(job_id, safe="") - response = requests.post( + response = _batch_processing_request( + requests.post, _batch_processing_jobs_url(workspace_url, f"/{encoded}/abort"), headers=_batch_processing_headers(api_key), json={}, @@ -1823,7 +1835,8 @@ def abort_batch_processing_job(api_key, workspace_url, job_id): def restart_batch_processing_job(api_key, workspace_url, job_id): """Restart one Batch Processing job with its existing configuration.""" encoded = quote(job_id, safe="") - response = requests.post( + response = _batch_processing_request( + requests.post, _batch_processing_jobs_url(workspace_url, f"/{encoded}/restart"), headers=_batch_processing_headers(api_key), json={}, diff --git a/tests/adapters/test_rfapi_batch_processing.py b/tests/adapters/test_rfapi_batch_processing.py index 9415fd6c..ee42ed97 100644 --- a/tests/adapters/test_rfapi_batch_processing.py +++ b/tests/adapters/test_rfapi_batch_processing.py @@ -5,6 +5,8 @@ import unittest from unittest.mock import Mock, patch +from requests.exceptions import ConnectionError, Timeout + from roboflow.adapters import rfapi @@ -71,6 +73,25 @@ def test_error_preserves_http_status_for_cli_exit_codes(self, mock_get) -> None: self.assertEqual(ctx.exception.status_code, 404) self.assertEqual(str(ctx.exception), "Job not found") + def test_transport_errors_are_translated_for_cli_recovery(self) -> None: + for transport_error in (ConnectionError("connection reset"), Timeout("request timed out")): + with self.subTest(transport_error=type(transport_error).__name__): + with patch( + "roboflow.adapters.rfapi.requests.post", + side_effect=transport_error, + ): + with self.assertRaises(rfapi.RoboflowError) as ctx: + rfapi.create_asset_library_batch_job( + "private-key", + "workspace-1", + workflow_id="workflow-1", + idempotency_key="request-123", + image_ids=["image-1"], + ) + + self.assertIs(ctx.exception.__cause__, transport_error) + self.assertEqual(str(ctx.exception), str(transport_error)) + if __name__ == "__main__": unittest.main() diff --git a/tests/cli/test_batch_handler.py b/tests/cli/test_batch_handler.py index bc605489..fca7e064 100644 --- a/tests/cli/test_batch_handler.py +++ b/tests/cli/test_batch_handler.py @@ -6,6 +6,7 @@ import unittest from unittest.mock import patch +from requests.exceptions import ConnectionError, Timeout from typer.testing import CliRunner from roboflow.cli import app @@ -127,6 +128,23 @@ def test_unsafe_request_id_fails_before_network(self, mock_create) -> None: self.assertIn("--request-id must be", result.output) mock_create.assert_not_called() + def test_ambiguous_transport_failure_preserves_generated_request_id(self) -> None: + request_id = "generated-request-123" + for transport_error in (ConnectionError("connection reset"), Timeout("request timed out")): + with self.subTest(transport_error=type(transport_error).__name__): + with ( + patch("roboflow.cli.handlers.batch.uuid.uuid4", return_value=request_id), + patch("roboflow.adapters.rfapi.requests.post", side_effect=transport_error), + ): + result = runner.invoke( + app, + [*BASE, "batch", "create", "--workflow", "workflow-1", "--all"], + ) + + self.assertEqual(result.exit_code, 1, result.output) + self.assertIn(str(transport_error), result.output) + self.assertIn(f"--request-id {request_id}", result.output) + class TestBatchLifecycle(unittest.TestCase): @patch("roboflow.adapters.rfapi.get_batch_processing_job") From d97b26fef545b093791e3505d3ecdde1dd4669ad Mon Sep 17 00:00:00 2001 From: Leo Ueno Date: Tue, 1 Sep 2026 18:23:51 -0700 Subject: [PATCH 4/4] Harden Batch Processing CLI requests --- roboflow/adapters/rfapi.py | 8 ++++- roboflow/cli/handlers/batch.py | 36 +++++++++---------- tests/adapters/test_rfapi_batch_processing.py | 1 + tests/cli/test_batch_handler.py | 6 +++- 4 files changed, 31 insertions(+), 20 deletions(-) diff --git a/roboflow/adapters/rfapi.py b/roboflow/adapters/rfapi.py index 1b213725..9e9b4fc2 100644 --- a/roboflow/adapters/rfapi.py +++ b/roboflow/adapters/rfapi.py @@ -1714,6 +1714,8 @@ def get_video_job_status(api_key, job_id): # Batch Processing (Asset Library orchestration) # --------------------------------------------------------------------------- +BATCH_PROCESSING_REQUEST_TIMEOUT = (10, 60) + def _batch_processing_jobs_url(workspace_url, suffix=""): return f"{API_URL}/batch-processing/v1/external/{workspace_url}/jobs{suffix}" @@ -1724,7 +1726,6 @@ def _asset_library_batch_processing_url(workspace_url): def _batch_processing_headers(api_key): - # Keep credentials out of URLs, proxy logs, and shell history. validateToken supports Bearer. return {"Authorization": f"Bearer {api_key}"} @@ -1781,6 +1782,7 @@ def create_asset_library_batch_job( _asset_library_batch_processing_url(workspace_url), headers=_batch_processing_headers(api_key), json=payload, + timeout=BATCH_PROCESSING_REQUEST_TIMEOUT, ) if response.status_code != 202: _raise_for_batch_processing_response(response) @@ -1799,6 +1801,7 @@ def list_batch_processing_jobs(api_key, workspace_url, *, page_size=10, next_pag _batch_processing_jobs_url(workspace_url), headers=_batch_processing_headers(api_key), params=params, + timeout=BATCH_PROCESSING_REQUEST_TIMEOUT, ) if response.status_code != 200: _raise_for_batch_processing_response(response) @@ -1812,6 +1815,7 @@ def get_batch_processing_job(api_key, workspace_url, job_id): requests.get, _batch_processing_jobs_url(workspace_url, f"/{encoded}"), headers=_batch_processing_headers(api_key), + timeout=BATCH_PROCESSING_REQUEST_TIMEOUT, ) if response.status_code != 200: _raise_for_batch_processing_response(response) @@ -1826,6 +1830,7 @@ def abort_batch_processing_job(api_key, workspace_url, job_id): _batch_processing_jobs_url(workspace_url, f"/{encoded}/abort"), headers=_batch_processing_headers(api_key), json={}, + timeout=BATCH_PROCESSING_REQUEST_TIMEOUT, ) if response.status_code != 200: _raise_for_batch_processing_response(response) @@ -1840,6 +1845,7 @@ def restart_batch_processing_job(api_key, workspace_url, job_id): _batch_processing_jobs_url(workspace_url, f"/{encoded}/restart"), headers=_batch_processing_headers(api_key), json={}, + timeout=BATCH_PROCESSING_REQUEST_TIMEOUT, ) if response.status_code != 200: _raise_for_batch_processing_response(response) diff --git a/roboflow/cli/handlers/batch.py b/roboflow/cli/handlers/batch.py index dfd29940..08a4bb5c 100644 --- a/roboflow/cli/handlers/batch.py +++ b/roboflow/cli/handlers/batch.py @@ -31,7 +31,7 @@ def create( ] = None, query: Annotated[ Optional[str], - typer.Option(help="Reviewed structured RoboQL filter selecting all current matches"), + typer.Option(help="RoboQL filter selecting all current matches"), ] = None, all_images: Annotated[ bool, @@ -182,11 +182,11 @@ def _create(args) -> None: # noqa: ANN001 result = {**result, "requestId": request_id} text = ( - f"Queued {result.get('displayName') or result.get('jobId')}\n" - f"jobId={result.get('jobId')}\n" - f"taskId={result.get('taskId')}\n" + f"Queued {result['displayName']}\n" + f"jobId={result['jobId']}\n" + f"taskId={result['taskId']}\n" f"requestId={request_id}\n" - f"Next: roboflow asynctasks wait {result.get('taskId')}" + f"Next: roboflow asynctasks wait {result['taskId']}" ) output(args, result, text=text) @@ -207,19 +207,19 @@ def _status(args) -> None: # noqa: ANN001 args, exc, auth_hint="Check the API key has 'batch-processing:read' scope.", - not_found_hint="Check the job ID and workspace.", + not_found_hint=( + "If the job was just queued, wait using the task ID returned by 'batch create'; " + "otherwise check the job ID and workspace." + ), ) return - job = result.get("job", {}) - state = job.get("currentStage") or ("terminal" if job.get("isTerminal") else "queued") + job = result["job"] + state = job.get("currentStage") or ("terminal" if job["isTerminal"] else "queued") output( args, result, - text=( - f"jobId={job.get('jobId', args.job_id)} state={state} " - f"terminal={job.get('isTerminal')} error={job.get('error')}" - ), + text=(f"jobId={job['jobId']} state={state} terminal={job['isTerminal']} error={job['error']}"), ) @@ -251,13 +251,13 @@ def _list(args) -> None: # noqa: ANN001 rows = [ { - "jobId": job.get("jobId", ""), - "name": job.get("name", ""), - "stage": job.get("currentStage") or ("terminal" if job.get("isTerminal") else "queued"), - "error": job.get("error", False), - "updated": job.get("lastUpdate", ""), + "jobId": job["jobId"], + "name": job["name"], + "stage": job.get("currentStage") or ("terminal" if job["isTerminal"] else "queued"), + "error": job["error"], + "updated": job["lastUpdate"], } - for job in result.get("jobs", []) + for job in result["jobs"] ] table = format_table(rows, columns=["jobId", "name", "stage", "error", "updated"]) if result.get("nextPageToken"): diff --git a/tests/adapters/test_rfapi_batch_processing.py b/tests/adapters/test_rfapi_batch_processing.py index ee42ed97..5e741299 100644 --- a/tests/adapters/test_rfapi_batch_processing.py +++ b/tests/adapters/test_rfapi_batch_processing.py @@ -34,6 +34,7 @@ def test_create_uses_bearer_auth_and_idempotent_payload(self, mock_post) -> None f"{rfapi.API_URL}/batch-processing/v1/external/workspace-1/asset-library/jobs", ) self.assertEqual(kwargs["json"]["idempotencyKey"], "request-123") + self.assertEqual(kwargs["timeout"], rfapi.BATCH_PROCESSING_REQUEST_TIMEOUT) @patch("roboflow.adapters.rfapi.requests.get") def test_list_uses_canonical_jobs_endpoint(self, mock_get) -> None: diff --git a/tests/cli/test_batch_handler.py b/tests/cli/test_batch_handler.py index fca7e064..c900084e 100644 --- a/tests/cli/test_batch_handler.py +++ b/tests/cli/test_batch_handler.py @@ -74,7 +74,11 @@ def test_create_exact_selection_has_stable_machine_output(self, mock_create) -> @patch("roboflow.adapters.rfapi.create_asset_library_batch_job") def test_all_is_explicit_empty_query_not_an_omitted_selection(self, mock_create) -> None: - mock_create.return_value = {"taskId": "task-1", "jobId": "al-123"} + mock_create.return_value = { + "taskId": "task-1", + "jobId": "al-123", + "displayName": "Asset Library Batch", + } result = runner.invoke( app,