Skip to content
Draft
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
21 changes: 21 additions & 0 deletions CLI-COMMANDS.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 <job-id>
roboflow batch list
roboflow batch abort <job-id>
roboflow batch restart <job-id>
```

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
Expand Down
119 changes: 119 additions & 0 deletions roboflow/adapters/rfapi.py
Original file line number Diff line number Diff line change
Expand Up @@ -1376,6 +1376,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
# ---------------------------------------------------------------------------
Expand Down
2 changes: 1 addition & 1 deletion roboflow/cli/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
Loading
Loading