diff --git a/CLI-COMMANDS.md b/CLI-COMMANDS.md index 4593b802..7fcadfe4 100644 --- a/CLI-COMMANDS.md +++ b/CLI-COMMANDS.md @@ -170,9 +170,50 @@ roboflow folder delete ```bash roboflow annotation batch list -p my-project roboflow annotation batch get -p my-project -roboflow annotation job list -p my-project +roboflow annotation batch admin-list -p my-project --limit 50 +roboflow annotation batch images -p my-project +roboflow annotation batch create -p my-project --source-batch-id \ + --image-id --name "Review batch" +roboflow annotation batch merge -p my-project --source-batch-id \ + --target-batch-id --yes + +roboflow annotation job admin-list -p my-project --limit 50 +roboflow annotation job admin-get -p my-project +roboflow annotation job admin-create -p my-project --batch \ + --labeler a@co.com --reviewer b@co.com --name "Label round 1" roboflow annotation job create -p my-project --name "Label round 1" \ --batch --num-images 100 --labeler a@co.com --reviewer b@co.com +roboflow annotation job images -p my-project +roboflow annotation job submit-review -p my-project +roboflow annotation job review-image -p my-project --status approved +roboflow annotation job return-edits -p my-project --new-labeler a@co.com +roboflow annotation job accept -p my-project --split-method split \ + --status approved --train-count 80 --valid-count 10 --test-count 10 --yes +``` + +Use `admin-list --after ` to fetch the next page. Commands +that delete or consolidate resources, or accept images into Dataset, require +`--yes` when run non-interactively. Run a batch or job subcommand with `--help` +for the full administration surface. + +The same operations are available from a `Project` in Python: + +```python +batches = project.get_annotation_batches(limit=50) +job = project.create_annotation_job_admin( + batch_id="batch-id", + labeler_email="labeler@example.com", + reviewer_email="reviewer@example.com", +) +project.submit_annotation_job_for_review(job["id"]) +project.accept_annotation_job_images( + job["id"], + split_method="split", + statuses_to_include=["approved"], + train_count=80, + valid_count=10, + test_count=10, +) ``` ### RFDM devices (v2 deployments) diff --git a/roboflow/adapters/rfapi.py b/roboflow/adapters/rfapi.py index e2631122..43cfeb6a 100644 --- a/roboflow/adapters/rfapi.py +++ b/roboflow/adapters/rfapi.py @@ -936,44 +936,128 @@ def get_zip_upload_status(api_key, workspace_url, task_id) -> dict: # --------------------------------------------------------------------------- -# Phase 2: Annotation batch & job endpoints +# Annotation batch & job endpoints # --------------------------------------------------------------------------- def list_batches(api_key, workspace_url, project_url): - """GET /{ws}/{proj}/batches — list annotation batches.""" + """GET /{ws}/{proj}/batches — list established upload batches.""" response = requests.get(f"{API_URL}/{workspace_url}/{project_url}/batches", params={"api_key": api_key}) - if response.status_code != 200: - raise RoboflowError(response.text) - return response.json() + return _annotation_administration_response(response) def get_batch(api_key, workspace_url, project_url, batch_id): - """GET /{ws}/{proj}/batches/{batch_id} — get batch details.""" + """GET /{ws}/{proj}/batches/{batch_id} — get an established upload batch.""" response = requests.get(f"{API_URL}/{workspace_url}/{project_url}/batches/{batch_id}", params={"api_key": api_key}) - if response.status_code != 200: - raise RoboflowError(response.text) - return response.json() + return _annotation_administration_response(response) + + +def list_annotation_batches(api_key, workspace_url, project_url, *, limit=50, after=None, show_empty=False): + """List annotation-board batches with cursor pagination.""" + response = requests.get( + f"{API_URL}/{workspace_url}/{project_url}/annotation-batches", + params=_annotation_pagination_params(api_key, limit=limit, after=after, show_empty=show_empty), + ) + return _annotation_administration_response(response) + + +def get_annotation_batch(api_key, workspace_url, project_url, batch_id): + """Get one annotation-board batch.""" + response = requests.get( + f"{API_URL}/{workspace_url}/{project_url}/annotation-batches/{batch_id}", + params={"api_key": api_key}, + ) + return _annotation_administration_response(response) + + +def list_annotation_batch_images(api_key, workspace_url, project_url, batch_id, *, limit=50, after=None): + """List image IDs in an annotation batch with cursor pagination.""" + response = requests.get( + f"{API_URL}/{workspace_url}/{project_url}/annotation-batches/{batch_id}/images", + params=_annotation_pagination_params(api_key, limit=limit, after=after), + ) + return _annotation_administration_response(response) + + +def create_annotation_batch(api_key, workspace_url, project_url, *, source_batch_id, image_ids, name=None): + """Move selected images from one batch into a new annotation batch.""" + payload = {"sourceBatchId": source_batch_id, "imageIds": image_ids} + if name: + payload["name"] = name + response = requests.post( + f"{API_URL}/{workspace_url}/{project_url}/annotation-batches", + params={"api_key": api_key}, + json=payload, + ) + return _annotation_administration_response(response) + + +def merge_annotation_batches(api_key, workspace_url, project_url, *, source_batch_ids, target_batch_id): + """Move source-batch images into a target batch and remove the emptied sources.""" + response = requests.post( + f"{API_URL}/{workspace_url}/{project_url}/annotation-batches/merge", + params={"api_key": api_key}, + json={"sourceBatchIds": source_batch_ids, "targetBatchId": target_batch_id}, + ) + return _annotation_administration_response(response) + + +def delete_annotation_batch(api_key, workspace_url, project_url, batch_id, *, permanent=False): + """Delete a batch, retaining its images as unassigned unless permanent is true.""" + response = requests.delete( + f"{API_URL}/{workspace_url}/{project_url}/annotation-batches/{batch_id}", + params={"api_key": api_key, "permanent": str(permanent).lower()}, + ) + return _annotation_administration_response(response) def list_annotation_jobs(api_key, workspace_url, project_url): - """GET /{ws}/{proj}/jobs — list annotation jobs.""" - response = requests.get(f"{API_URL}/{workspace_url}/{project_url}/jobs", params={"api_key": api_key}) - if response.status_code != 200: - raise RoboflowError(response.text) - return response.json() + """List all annotation jobs through the established jobs endpoint.""" + response = requests.get( + f"{API_URL}/{workspace_url}/{project_url}/jobs", + params={"api_key": api_key}, + ) + return _annotation_administration_response(response) + + +def list_annotation_jobs_admin(api_key, workspace_url, project_url, *, limit=50, after=None, show_empty=False): + """List annotation jobs through the administration endpoint with cursor pagination.""" + response = requests.get( + f"{API_URL}/{workspace_url}/{project_url}/annotation-jobs", + params=_annotation_pagination_params(api_key, limit=limit, after=after, show_empty=show_empty), + ) + return _annotation_administration_response(response) def get_annotation_job(api_key, workspace_url, project_url, job_id): - """GET /{ws}/{proj}/jobs/{job_id} — get annotation job details.""" - response = requests.get(f"{API_URL}/{workspace_url}/{project_url}/jobs/{job_id}", params={"api_key": api_key}) - if response.status_code != 200: - raise RoboflowError(response.text) - return response.json() + """Get one annotation job through the established jobs endpoint.""" + response = requests.get( + f"{API_URL}/{workspace_url}/{project_url}/jobs/{job_id}", + params={"api_key": api_key}, + ) + return _annotation_administration_response(response) + + +def get_annotation_job_admin(api_key, workspace_url, project_url, job_id): + """Get one annotation job through the administration endpoint.""" + response = requests.get( + f"{API_URL}/{workspace_url}/{project_url}/annotation-jobs/{job_id}", + params={"api_key": api_key}, + ) + return _annotation_administration_response(response) + + +def list_annotation_job_images(api_key, workspace_url, project_url, job_id, *, limit=50, after=None): + """List image IDs assigned to a job with cursor pagination.""" + response = requests.get( + f"{API_URL}/{workspace_url}/{project_url}/annotation-jobs/{job_id}/images", + params=_annotation_pagination_params(api_key, limit=limit, after=after), + ) + return _annotation_administration_response(response) def create_annotation_job(api_key, workspace_url, project_url, *, name, batch_id=None, assignees=None): - """POST /{ws}/{proj}/jobs — create an annotation job.""" + """Create an annotation job with the established low-level adapter contract.""" payload = {"name": name} if batch_id: payload["batchId"] = batch_id @@ -984,8 +1068,256 @@ def create_annotation_job(api_key, workspace_url, project_url, *, name, batch_id params={"api_key": api_key}, json=payload, ) - if response.status_code not in (200, 201): - raise RoboflowError(response.text) + return _annotation_administration_response(response) + + +def create_annotation_job_from_batch( + api_key, + workspace_url, + project_url, + *, + batch_id, + labeler_email, + reviewer_email, + name=None, + num_images=None, + instructions=None, +): + """Create a fully assigned job through the established jobs endpoint.""" + payload = { + "name": name, + "batch": batch_id, + "num_images": num_images, + "labelerEmail": labeler_email, + "reviewerEmail": reviewer_email, + } + if instructions is not None: + payload["instructionText"] = instructions + response = requests.post( + f"{API_URL}/{workspace_url}/{project_url}/jobs", + params={"api_key": api_key}, + json=payload, + ) + return _annotation_administration_response(response) + + +def create_annotation_job_admin( + api_key, + workspace_url, + project_url, + *, + batch_id, + labeler_email, + reviewer_email, + name=None, + num_images=None, + instructions=None, +): + """Create a job through the administration endpoint.""" + payload = { + "batchId": batch_id, + "labelerEmail": labeler_email, + "reviewerEmail": reviewer_email, + } + payload.update( + { + key: value + for key, value in {"name": name, "numImages": num_images, "instructions": instructions}.items() + if value is not None + } + ) + response = requests.post( + f"{API_URL}/{workspace_url}/{project_url}/annotation-jobs", + params={"api_key": api_key}, + json=payload, + ) + return _annotation_administration_response(response) + + +def reassign_annotation_job_images( + api_key, + workspace_url, + project_url, + *, + image_ids, + labeler_email, + reviewer_email=None, + instructions=None, + name=None, +): + """Create a job by removing selected images from their prior assignment.""" + payload = {"imageIds": image_ids, "labelerEmail": labeler_email} + payload.update( + { + key: value + for key, value in { + "reviewerEmail": reviewer_email, + "instructions": instructions, + "name": name, + }.items() + if value is not None + } + ) + response = requests.post( + f"{API_URL}/{workspace_url}/{project_url}/annotation-jobs/reassign-images", + params={"api_key": api_key}, + json=payload, + ) + return _annotation_administration_response(response) + + +def add_images_to_annotation_job(api_key, workspace_url, project_url, job_id, *, image_ids): + """Move selected images into an existing annotation job.""" + response = requests.post( + f"{API_URL}/{workspace_url}/{project_url}/annotation-jobs/{job_id}/images", + params={"api_key": api_key}, + json={"imageIds": image_ids}, + ) + return _annotation_administration_response(response) + + +def update_annotation_job( + api_key, + workspace_url, + project_url, + job_id, + *, + labeler_email=None, + reviewer_email=None, + instructions=None, +): + """Update exactly one of a job's labeler, reviewer, or instructions.""" + payload = { + key: value + for key, value in { + "labelerEmail": labeler_email, + "reviewerEmail": reviewer_email, + "instructions": instructions, + }.items() + if value is not None + } + if len(payload) != 1: + raise ValueError("Provide exactly one of labeler_email, reviewer_email, or instructions") + response = requests.patch( + f"{API_URL}/{workspace_url}/{project_url}/annotation-jobs/{job_id}", + params={"api_key": api_key}, + json=payload, + ) + return _annotation_administration_response(response) + + +def submit_annotation_job_for_review(api_key, workspace_url, project_url, job_id): + """Advance a labeling job into review.""" + response = requests.post( + f"{API_URL}/{workspace_url}/{project_url}/annotation-jobs/{job_id}/submit-review", + params={"api_key": api_key}, + json={}, + ) + return _annotation_administration_response(response) + + +def return_annotation_job_for_edits(api_key, workspace_url, project_url, job_id, *, new_labeler_email=None): + """Move a review job back to labeling, optionally with a new labeler.""" + payload = {"newLabelerEmail": new_labeler_email} if new_labeler_email else {} + response = requests.post( + f"{API_URL}/{workspace_url}/{project_url}/annotation-jobs/{job_id}/return-edits", + params={"api_key": api_key}, + json=payload, + ) + return _annotation_administration_response(response) + + +def review_annotation_job_image(api_key, workspace_url, project_url, job_id, image_id, *, status): + """Set the review status for one image in a job.""" + response = requests.post( + f"{API_URL}/{workspace_url}/{project_url}/annotation-jobs/{job_id}/images/{image_id}/status", + params={"api_key": api_key}, + json={"status": status}, + ) + return _annotation_administration_response(response) + + +def review_annotation_job_images(api_key, workspace_url, project_url, job_id, *, status, current_status): + """Set a status for every job image matching the supplied current status.""" + response = requests.post( + f"{API_URL}/{workspace_url}/{project_url}/annotation-jobs/{job_id}/images/status", + params={"api_key": api_key}, + json={"status": status, "currentStatus": current_status}, + ) + return _annotation_administration_response(response) + + +def accept_annotation_job_images( + api_key, + workspace_url, + project_url, + job_id, + *, + split_method, + statuses_to_include, + train_count, + valid_count, + test_count, + image_ids=None, +): + """Accept selected job images into Dataset and assign their splits.""" + payload = { + "splitMethod": split_method, + "statusesToInclude": statuses_to_include, + "trainCount": train_count, + "validCount": valid_count, + "testCount": test_count, + } + if image_ids is not None: + payload["imageIds"] = image_ids + response = requests.post( + f"{API_URL}/{workspace_url}/{project_url}/annotation-jobs/{job_id}/accept", + params={"api_key": api_key}, + json=payload, + ) + return _annotation_administration_response(response) + + +def move_annotation_job_to_unassigned(api_key, workspace_url, project_url, job_id): + """Remove a job while retaining its images in an unassigned batch.""" + response = requests.post( + f"{API_URL}/{workspace_url}/{project_url}/annotation-jobs/{job_id}/move-to-unassigned", + params={"api_key": api_key}, + json={}, + ) + return _annotation_administration_response(response) + + +def delete_annotation_job_annotations(api_key, workspace_url, project_url, job_id): + """Delete project annotations from all images assigned to a job.""" + response = requests.delete( + f"{API_URL}/{workspace_url}/{project_url}/annotation-jobs/{job_id}/annotations", + params={"api_key": api_key}, + ) + return _annotation_administration_response(response) + + +def _annotation_pagination_params(api_key, *, limit, after=None, show_empty=None): + params = {"api_key": api_key, "limit": limit} + if after: + params["after"] = after + if show_empty is not None: + params["showEmpty"] = str(show_empty).lower() + return params + + +def _annotation_administration_response(response): + if not 200 <= response.status_code < 300: + message = response.text + try: + error = response.json().get("error") + if isinstance(error, dict): + message = error.get("message", str(error)) + elif error: + message = str(error) + except (AttributeError, ValueError): + pass + raise RoboflowError(message, status_code=response.status_code) return response.json() diff --git a/roboflow/cli/handlers/annotation.py b/roboflow/cli/handlers/annotation.py index 2259960e..de5b7dfd 100644 --- a/roboflow/cli/handlers/annotation.py +++ b/roboflow/cli/handlers/annotation.py @@ -2,7 +2,8 @@ from __future__ import annotations -from typing import Annotated +import json +from typing import Annotated, Any, Callable, Optional import typer @@ -11,24 +12,17 @@ annotation_app = typer.Typer(cls=SortedGroup, help="Annotation management commands", no_args_is_help=True) batch_app = typer.Typer(cls=SortedGroup, help="Annotation batch commands", no_args_is_help=True) job_app = typer.Typer(cls=SortedGroup, help="Annotation job commands", no_args_is_help=True) - annotation_app.add_typer(batch_app, name="batch") annotation_app.add_typer(job_app, name="job") -# --------------------------------------------------------------------------- -# batch commands -# --------------------------------------------------------------------------- - - @batch_app.command("list") def batch_list( ctx: typer.Context, project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], ) -> None: - """List annotation batches.""" - args = ctx_to_args(ctx, project=project) - _batch_list(args) + """List established upload batches.""" + _batch_list(ctx_to_args(ctx, project=project)) @batch_app.command("get") @@ -37,14 +31,115 @@ def batch_get( batch_id: Annotated[str, typer.Argument(help="Batch ID")], project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], ) -> None: - """Get annotation batch details.""" - args = ctx_to_args(ctx, batch_id=batch_id, project=project) - _batch_get(args) + """Get an established upload batch.""" + _simple_command(ctx_to_args(ctx, batch_id=batch_id, project=project), "get_batch", batch_id) + + +@batch_app.command("admin-list") +def batch_admin_list( + ctx: typer.Context, + project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], + limit: Annotated[int, typer.Option(help="Maximum batches to return, from 1 to 200")] = 50, + after: Annotated[Optional[str], typer.Option(help="Continuation token from the previous page")] = None, + show_empty: Annotated[bool, typer.Option("--show-empty", help="Include batches with no images")] = False, +) -> None: + """List annotation-board batches with cursor pagination.""" + _simple_command( + ctx_to_args(ctx, project=project), + "list_annotation_batches", + limit=limit, + after=after, + show_empty=show_empty, + ) + + +@batch_app.command("admin-get") +def batch_admin_get( + ctx: typer.Context, + batch_id: Annotated[str, typer.Argument(help="Batch ID")], + project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], +) -> None: + """Get an annotation-board batch.""" + _simple_command(ctx_to_args(ctx, project=project), "get_annotation_batch", batch_id) + + +@batch_app.command("images") +def batch_images( + ctx: typer.Context, + batch_id: Annotated[str, typer.Argument(help="Batch ID")], + project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], + limit: Annotated[int, typer.Option(help="Maximum image IDs to return, from 1 to 200")] = 50, + after: Annotated[Optional[str], typer.Option(help="Continuation token from the previous page")] = None, +) -> None: + """List image IDs in an annotation batch.""" + _simple_command( + ctx_to_args(ctx, project=project), + "list_annotation_batch_images", + batch_id, + limit=limit, + after=after, + ) + + +@batch_app.command("create") +def batch_create( + ctx: typer.Context, + project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], + source_batch_id: Annotated[str, typer.Option("--source-batch-id", help="Source batch ID")], + image_ids: Annotated[list[str], typer.Option("--image-id", help="Image ID to move; repeat for multiple images")], + name: Annotated[Optional[str], typer.Option(help="Optional new batch name")] = None, +) -> None: + """Move selected images into a new annotation batch.""" + _simple_command( + ctx_to_args(ctx, project=project), + "create_annotation_batch", + source_batch_id=source_batch_id, + image_ids=image_ids, + name=name, + success="Created annotation batch.", + ) + + +@batch_app.command("merge") +def batch_merge( + ctx: typer.Context, + project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], + source_batch_ids: Annotated[ + list[str], typer.Option("--source-batch-id", help="Source batch ID; repeat for multiple batches") + ], + target_batch_id: Annotated[str, typer.Option("--target-batch-id", help="Target batch ID")], + yes: Annotated[bool, typer.Option("-y", "--yes", help="Confirm the merge")] = False, +) -> None: + """Merge source batches into a target batch.""" + args = ctx_to_args(ctx, project=project, yes=yes) + if _confirm(args, "Merge the source batches and remove the emptied batches?"): + _simple_command( + args, + "merge_annotation_batches", + source_batch_ids=source_batch_ids, + target_batch_id=target_batch_id, + success="Merged annotation batches.", + ) -# --------------------------------------------------------------------------- -# job commands -# --------------------------------------------------------------------------- +@batch_app.command("delete") +def batch_delete( + ctx: typer.Context, + batch_id: Annotated[str, typer.Argument(help="Batch ID")], + project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], + permanent: Annotated[bool, typer.Option(help="Also delete the batch's image sources")] = False, + yes: Annotated[bool, typer.Option("-y", "--yes", help="Confirm deletion")] = False, +) -> None: + """Delete an annotation batch.""" + args = ctx_to_args(ctx, project=project, yes=yes) + if _confirm(args, "Delete this annotation batch?"): + _simple_command( + args, + "delete_annotation_batch", + batch_id, + permanent=permanent, + success="Deleted annotation batch.", + ) @job_app.command("list") @@ -52,107 +147,372 @@ def job_list( ctx: typer.Context, project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], ) -> None: - """List annotation jobs.""" - args = ctx_to_args(ctx, project=project) - _job_list(args) + """List annotation jobs with legacy list-only output.""" + _job_list(ctx_to_args(ctx, project=project)) + + +@job_app.command("admin-list") +def job_admin_list( + ctx: typer.Context, + project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], + limit: Annotated[int, typer.Option(help="Maximum jobs to return, from 1 to 200")] = 50, + after: Annotated[Optional[str], typer.Option(help="Continuation token from the previous page")] = None, + show_empty: Annotated[bool, typer.Option("--show-empty", help="Include jobs with no images")] = False, +) -> None: + """List annotation jobs with the full paginated response.""" + _simple_command( + ctx_to_args(ctx, project=project), + "list_annotation_jobs_admin", + limit=limit, + after=after, + show_empty=show_empty, + ) @job_app.command("get") def job_get( ctx: typer.Context, - job_id: Annotated[str, typer.Argument(help="Job ID")], + job_id: Annotated[str, typer.Argument(help="Annotation job ID")], project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], ) -> None: """Get annotation job details.""" - args = ctx_to_args(ctx, job_id=job_id, project=project) - _job_get(args) + _simple_command(ctx_to_args(ctx, project=project), "get_annotation_job", job_id) + + +@job_app.command("admin-get") +def job_admin_get( + ctx: typer.Context, + job_id: Annotated[str, typer.Argument(help="Annotation job ID")], + project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], +) -> None: + """Get an annotation job through the administration endpoint.""" + _simple_command(ctx_to_args(ctx, project=project), "get_annotation_job_admin", job_id) + + +@job_app.command("images") +def job_images( + ctx: typer.Context, + job_id: Annotated[str, typer.Argument(help="Annotation job ID")], + project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], + limit: Annotated[int, typer.Option(help="Maximum image IDs to return, from 1 to 200")] = 50, + after: Annotated[Optional[str], typer.Option(help="Continuation token from the previous page")] = None, +) -> None: + """List image IDs assigned to a job.""" + _simple_command( + ctx_to_args(ctx, project=project), + "list_annotation_job_images", + job_id, + limit=limit, + after=after, + ) @job_app.command("create") def job_create( ctx: typer.Context, project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], - name: Annotated[str, typer.Option(help="Job name")], - batch: Annotated[str, typer.Option(help="Batch ID")], - num_images: Annotated[int, typer.Option("--num-images", help="Number of images")], + batch: Annotated[str, typer.Option(help="Source batch ID")], labeler: Annotated[str, typer.Option(help="Labeler email")], reviewer: Annotated[str, typer.Option(help="Reviewer email")], + name: Annotated[Optional[str], typer.Option(help="Optional job name")] = None, + num_images: Annotated[Optional[int], typer.Option("--num-images", help="Number of images")] = None, + instructions: Annotated[Optional[str], typer.Option(help="Labeling instructions")] = None, ) -> None: - """Create an annotation job.""" - args = ctx_to_args( - ctx, - project=project, + """Create an annotation job from a batch.""" + _simple_command( + ctx_to_args(ctx, project=project), + "create_annotation_job_from_batch", + batch_id=batch, + labeler_email=labeler, + reviewer_email=reviewer, name=name, - batch=batch, num_images=num_images, - labeler=labeler, - reviewer=reviewer, + instructions=instructions, + success=f"Created annotation job{f': {name}' if name else '.'}", ) - _job_create(args) -# --------------------------------------------------------------------------- -# helpers -# --------------------------------------------------------------------------- +@job_app.command("admin-create") +def job_admin_create( + ctx: typer.Context, + project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], + batch: Annotated[str, typer.Option(help="Source batch ID")], + labeler: Annotated[str, typer.Option(help="Labeler email")], + reviewer: Annotated[str, typer.Option(help="Reviewer email")], + name: Annotated[Optional[str], typer.Option(help="Optional job name")] = None, + num_images: Annotated[Optional[int], typer.Option("--num-images", help="Number of images")] = None, + instructions: Annotated[Optional[str], typer.Option(help="Labeling instructions")] = None, +) -> None: + """Create an annotation job through the administration endpoint.""" + _simple_command( + ctx_to_args(ctx, project=project), + "create_annotation_job_admin", + batch_id=batch, + labeler_email=labeler, + reviewer_email=reviewer, + name=name, + num_images=num_images, + instructions=instructions, + success=f"Created annotation job{f': {name}' if name else '.'}", + ) -def _normalize_timestamps(obj): # noqa: ANN001 - """Recursively convert Firestore timestamp dicts ({"_seconds": N, "_nanoseconds": N}) to ISO 8601 strings.""" - from datetime import datetime, timezone +@job_app.command("reassign-images") +def job_reassign_images( + ctx: typer.Context, + project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], + image_ids: Annotated[ + list[str], typer.Option("--image-id", help="Image ID to reassign; repeat for multiple images") + ], + labeler: Annotated[str, typer.Option(help="Labeler email")], + reviewer: Annotated[Optional[str], typer.Option(help="Reviewer email")] = None, + instructions: Annotated[Optional[str], typer.Option(help="Labeling instructions")] = None, + name: Annotated[Optional[str], typer.Option(help="Optional job name")] = None, +) -> None: + """Create a job by explicitly reassigning images.""" + _simple_command( + ctx_to_args(ctx, project=project), + "reassign_annotation_job_images", + image_ids=image_ids, + labeler_email=labeler, + reviewer_email=reviewer, + instructions=instructions, + name=name, + success="Reassigned images to a new annotation job.", + ) - if isinstance(obj, dict): - if "_seconds" in obj and "_nanoseconds" in obj and len(obj) == 2: - return datetime.fromtimestamp(obj["_seconds"], tz=timezone.utc).isoformat() - return {k: _normalize_timestamps(v) for k, v in obj.items()} - if isinstance(obj, list): - return [_normalize_timestamps(item) for item in obj] - return obj + +@job_app.command("add-images") +def job_add_images( + ctx: typer.Context, + job_id: Annotated[str, typer.Argument(help="Target annotation job ID")], + project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], + image_ids: Annotated[list[str], typer.Option("--image-id", help="Image ID to add; repeat for multiple images")], + yes: Annotated[bool, typer.Option("-y", "--yes", help="Confirm reassignment of the images")] = False, +) -> None: + """Move selected images into an existing annotation job.""" + args = ctx_to_args(ctx, project=project, yes=yes) + if _confirm(args, "Move these images out of their current assignments and into this job?"): + _simple_command( + args, + "add_images_to_annotation_job", + job_id, + image_ids=image_ids, + success="Added images to the annotation job.", + ) + + +@job_app.command("update") +def job_update( + ctx: typer.Context, + job_id: Annotated[str, typer.Argument(help="Annotation job ID")], + project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], + labeler: Annotated[Optional[str], typer.Option(help="New labeler email")] = None, + reviewer: Annotated[Optional[str], typer.Option(help="New reviewer email")] = None, + instructions: Annotated[Optional[str], typer.Option(help="Replacement instructions")] = None, +) -> None: + """Update exactly one assignment field on a job.""" + _simple_command( + ctx_to_args(ctx, project=project), + "update_annotation_job", + job_id, + labeler_email=labeler, + reviewer_email=reviewer, + instructions=instructions, + success="Updated annotation job.", + ) + + +@job_app.command("submit-review") +def job_submit_review( + ctx: typer.Context, + job_id: Annotated[str, typer.Argument(help="Annotation job ID")], + project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], +) -> None: + """Advance a labeling job into review.""" + _simple_command( + ctx_to_args(ctx, project=project), + "submit_annotation_job_for_review", + job_id, + success="Submitted annotation job for review.", + ) -def _resolve_project_context(args): # noqa: ANN001 - """Resolve workspace/project from -p flag and return (api_key, ws, proj) or call output_error.""" +@job_app.command("return-edits") +def job_return_edits( + ctx: typer.Context, + job_id: Annotated[str, typer.Argument(help="Annotation job ID")], + project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], + new_labeler: Annotated[Optional[str], typer.Option("--new-labeler", help="Replacement labeler")] = None, +) -> None: + """Move a review job back to labeling.""" + _simple_command( + ctx_to_args(ctx, project=project), + "return_annotation_job_for_edits", + job_id, + new_labeler_email=new_labeler, + success="Returned annotation job for edits.", + ) + + +@job_app.command("review-image") +def job_review_image( + ctx: typer.Context, + job_id: Annotated[str, typer.Argument(help="Annotation job ID")], + image_id: Annotated[str, typer.Argument(help="Image ID in the job")], + project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], + status: Annotated[str, typer.Option(help="approved, rejected, annotated, or unannotated")], +) -> None: + """Set the review status for one image.""" + _simple_command( + ctx_to_args(ctx, project=project), + "review_annotation_job_image", + job_id, + image_id, + status=status, + success=f"Set image review status to {status}.", + ) + + +@job_app.command("review-images") +def job_review_images( + ctx: typer.Context, + job_id: Annotated[str, typer.Argument(help="Annotation job ID")], + project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], + current_status: Annotated[str, typer.Option("--current-status", help="Only images in this status")], + status: Annotated[str, typer.Option(help="New status")], +) -> None: + """Set a status for every matching image in a job.""" + _simple_command( + ctx_to_args(ctx, project=project), + "review_annotation_job_images", + job_id, + status=status, + current_status=current_status, + success=f"Set matching image review statuses to {status}.", + ) + + +@job_app.command("accept") +def job_accept( + ctx: typer.Context, + job_id: Annotated[str, typer.Argument(help="Annotation job ID")], + project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], + split_method: Annotated[str, typer.Option("--split-method", help="preset, split, train, valid, or test")], + statuses: Annotated[list[str], typer.Option("--status", help="Status to accept; repeat as needed")], + train_count: Annotated[int, typer.Option("--train-count", help="Images assigned to train")], + valid_count: Annotated[int, typer.Option("--valid-count", help="Images assigned to validation")], + test_count: Annotated[int, typer.Option("--test-count", help="Images assigned to test")], + image_ids: Annotated[ + Optional[list[str]], typer.Option("--image-id", help="Optional image subset; repeat as needed") + ] = None, + yes: Annotated[bool, typer.Option("-y", "--yes", help="Confirm Dataset acceptance")] = False, +) -> None: + """Accept job images into Dataset and assign their splits.""" + args = ctx_to_args(ctx, project=project, yes=yes) + if _confirm(args, "Accept these annotation job images into Dataset?"): + _simple_command( + args, + "accept_annotation_job_images", + job_id, + split_method=split_method, + statuses_to_include=statuses, + train_count=train_count, + valid_count=valid_count, + test_count=test_count, + image_ids=image_ids, + success="Accepted annotation job images into Dataset.", + ) + + +@job_app.command("move-to-unassigned") +def job_move_to_unassigned( + ctx: typer.Context, + job_id: Annotated[str, typer.Argument(help="Annotation job ID")], + project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], + yes: Annotated[bool, typer.Option("-y", "--yes", help="Confirm removal of the job")] = False, +) -> None: + """Remove a job and retain its images as unassigned.""" + args = ctx_to_args(ctx, project=project, yes=yes) + if _confirm(args, "Remove this job and move its images to unassigned?"): + _simple_command( + args, + "move_annotation_job_to_unassigned", + job_id, + success="Moved annotation job images to unassigned.", + ) + + +@job_app.command("delete-annotations") +def job_delete_annotations( + ctx: typer.Context, + job_id: Annotated[str, typer.Argument(help="Annotation job ID")], + project: Annotated[str, typer.Option("-p", "--project", help="Project ID")], + yes: Annotated[bool, typer.Option("-y", "--yes", help="Confirm annotation deletion")] = False, +) -> None: + """Delete project annotations from every image assigned to a job.""" + args = ctx_to_args(ctx, project=project, yes=yes) + if _confirm(args, "Delete every project annotation assigned to this job?"): + _simple_command( + args, + "delete_annotation_job_annotations", + job_id, + success="Deleted annotation job annotations.", + ) + + +def _resolve_project_context(args: Any) -> Optional[tuple[str, str, str]]: from roboflow.cli._output import output_error from roboflow.cli._resolver import resolve_resource from roboflow.config import load_roboflow_api_key try: - workspace_url, project_slug, _version = resolve_resource(args.project, workspace_override=args.workspace) + workspace, project, _version = resolve_resource(args.project, workspace_override=args.workspace) except ValueError as exc: output_error(args, str(exc)) return None - - api_key = args.api_key or load_roboflow_api_key(workspace_url) + api_key = args.api_key or load_roboflow_api_key(workspace) 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 api_key, workspace, project - return api_key, workspace_url, project_slug +def _call(args: Any, operation: Callable[[str, str, str], Any]) -> Any: + from roboflow.adapters import rfapi + from roboflow.cli._output import output_api_error, output_error -# --------------------------------------------------------------------------- -# handler implementations -# --------------------------------------------------------------------------- + context = _resolve_project_context(args) + if context is None: + return None + try: + return _normalize_timestamps(operation(*context)) + except rfapi.RoboflowError as exc: + output_api_error(args, exc) + except ValueError as exc: + output_error(args, str(exc)) + return None -def _batch_list(args): # noqa: ANN001 +def _simple_command(args: Any, method: str, *positional: Any, success: Optional[str] = None, **kwargs: Any) -> None: from roboflow.adapters import rfapi - from roboflow.cli._output import output, output_error - from roboflow.cli._table import format_table + from roboflow.cli._output import output - ctx = _resolve_project_context(args) - if ctx is None: - return - api_key, workspace_url, project_slug = ctx + operation = getattr(rfapi, method) + data = _call(args, lambda key, workspace, project: operation(key, workspace, project, *positional, **kwargs)) + if data is not None: + output(args, data, text=success or json.dumps(data, indent=2, default=str)) - try: - data = rfapi.list_batches(api_key, workspace_url, project_slug) - except rfapi.RoboflowError as exc: - output_error(args, str(exc), exit_code=3) - return - batches = data if isinstance(data, list) else data.get("batches", data) - batches = _normalize_timestamps(batches) +def _batch_list(args: Any) -> None: + from roboflow.adapters import rfapi + from roboflow.cli._output import output + from roboflow.cli._table import format_table + data = _call(args, lambda key, workspace, project: rfapi.list_batches(key, workspace, project)) + if data is None: + return + batches = data if isinstance(data, list) else data.get("batches", data) table = format_table( batches if isinstance(batches, list) else [], columns=["name", "id", "status", "images"], @@ -161,52 +521,15 @@ def _batch_list(args): # noqa: ANN001 output(args, batches, text=table) -def _batch_get(args): # noqa: ANN001 - from roboflow.adapters import rfapi - from roboflow.cli._output import output, output_error - - ctx = _resolve_project_context(args) - if ctx is None: - return - api_key, workspace_url, project_slug = ctx - - try: - data = rfapi.get_batch(api_key, workspace_url, project_slug, args.batch_id) - except rfapi.RoboflowError as exc: - output_error(args, str(exc), exit_code=3) - return - - data = _normalize_timestamps(data) - batch = data.get("batch", data) if isinstance(data, dict) else data - - lines = [] - if isinstance(batch, dict): - for key, val in batch.items(): - lines.append(f" {key:16s} {val}") - text = "\n".join(lines) if lines else "(no batch details)" - - output(args, data, text=text) - - -def _job_list(args): # noqa: ANN001 +def _job_list(args: Any) -> None: from roboflow.adapters import rfapi - from roboflow.cli._output import output, output_error + from roboflow.cli._output import output from roboflow.cli._table import format_table - ctx = _resolve_project_context(args) - if ctx is None: - return - api_key, workspace_url, project_slug = ctx - - try: - data = rfapi.list_annotation_jobs(api_key, workspace_url, project_slug) - except rfapi.RoboflowError as exc: - output_error(args, str(exc), exit_code=3) + data = _call(args, lambda key, workspace, project: rfapi.list_annotation_jobs(key, workspace, project)) + if data is None: return - jobs = data if isinstance(data, list) else data.get("jobs", data) - jobs = _normalize_timestamps(jobs) - table = format_table( jobs if isinstance(jobs, list) else [], columns=["name", "id", "status", "assigned_to"], @@ -215,61 +538,19 @@ def _job_list(args): # noqa: ANN001 output(args, jobs, text=table) -def _job_get(args): # noqa: ANN001 - from roboflow.adapters import rfapi - from roboflow.cli._output import output, output_error +def _confirm(args: Any, prompt: str) -> bool: + from roboflow.cli._output import confirm_destructive - ctx = _resolve_project_context(args) - if ctx is None: - return - api_key, workspace_url, project_slug = ctx - - try: - data = rfapi.get_annotation_job(api_key, workspace_url, project_slug, args.job_id) - except rfapi.RoboflowError as exc: - output_error(args, str(exc), exit_code=3) - return - - data = _normalize_timestamps(data) - job = data.get("job", data) if isinstance(data, dict) else data - - lines = [] - if isinstance(job, dict): - for key, val in job.items(): - lines.append(f" {key:16s} {val}") - text = "\n".join(lines) if lines else "(no job details)" - - output(args, data, text=text) - - -def _job_create(args): # noqa: ANN001 - import roboflow - from roboflow.cli._output import output, output_error, suppress_sdk_output - - ctx = _resolve_project_context(args) - if ctx is None: - return - _api_key, workspace_url, project_slug = ctx + return confirm_destructive(args, prompt=prompt) - with suppress_sdk_output(args): - try: - rf = roboflow.Roboflow(api_key=_api_key) - workspace = rf.workspace(workspace_url) - project = workspace.project(project_slug) - except Exception as exc: - output_error(args, str(exc)) - return - try: - result = project.create_annotation_job( - name=args.name, - batch_id=args.batch, - num_images=args.num_images, - labeler_email=args.labeler, - reviewer_email=args.reviewer, - ) - except Exception as exc: - output_error(args, str(exc)) - return +def _normalize_timestamps(obj: Any) -> Any: + from datetime import datetime, timezone - output(args, result, text=f"Created annotation job: {args.name}") + if isinstance(obj, dict): + if "_seconds" in obj and "_nanoseconds" in obj and len(obj) == 2: + return datetime.fromtimestamp(obj["_seconds"], tz=timezone.utc).isoformat() + return {key: _normalize_timestamps(value) for key, value in obj.items()} + if isinstance(obj, list): + return [_normalize_timestamps(item) for item in obj] + return obj diff --git a/roboflow/core/project.py b/roboflow/core/project.py index 8d4aad75..a5bf990f 100644 --- a/roboflow/core/project.py +++ b/roboflow/core/project.py @@ -884,16 +884,78 @@ def image(self, image_id: str) -> Dict: return image_details - def get_annotation_jobs(self) -> Dict: - """Get a list of all annotation jobs in the project. + def get_annotation_batches(self, limit: int = 50, after: Optional[str] = None, show_empty: bool = False) -> Dict: + """List annotation-board batches with cursor pagination.""" + return rfapi.list_annotation_batches( + self.__api_key, + self.__workspace, + self.__project_name, + limit=limit, + after=after, + show_empty=show_empty, + ) - Returns: - Dict: A dictionary containing the list of annotation jobs. - """ - from roboflow.adapters import rfapi + def get_annotation_batch(self, batch_id: str) -> Dict: + """Get one annotation-board batch.""" + return rfapi.get_annotation_batch(self.__api_key, self.__workspace, self.__project_name, batch_id) + + def get_annotation_batch_images(self, batch_id: str, limit: int = 50, after: Optional[str] = None) -> Dict: + """List image IDs in an annotation batch with cursor pagination.""" + return rfapi.list_annotation_batch_images( + self.__api_key, + self.__workspace, + self.__project_name, + batch_id, + limit=limit, + after=after, + ) + + def create_annotation_batch(self, source_batch_id: str, image_ids: List[str], name: Optional[str] = None) -> Dict: + """Move selected images from one batch into a new annotation batch.""" + return rfapi.create_annotation_batch( + self.__api_key, + self.__workspace, + self.__project_name, + source_batch_id=source_batch_id, + image_ids=image_ids, + name=name, + ) + def merge_annotation_batches(self, source_batch_ids: List[str], target_batch_id: str) -> Dict: + """Move source-batch images into a target batch and remove the emptied sources.""" + return rfapi.merge_annotation_batches( + self.__api_key, + self.__workspace, + self.__project_name, + source_batch_ids=source_batch_ids, + target_batch_id=target_batch_id, + ) + + def delete_annotation_batch(self, batch_id: str, permanent: bool = False) -> Dict: + """Delete a batch, retaining its images as unassigned unless permanent is true.""" + return rfapi.delete_annotation_batch( + self.__api_key, + self.__workspace, + self.__project_name, + batch_id, + permanent=permanent, + ) + + def get_annotation_jobs(self) -> Dict: + """List all annotation jobs through the established jobs endpoint.""" return rfapi.list_annotation_jobs(self.__api_key, self.__workspace, self.__project_name) + def get_annotation_jobs_admin(self, limit: int = 50, after: Optional[str] = None, show_empty: bool = False) -> Dict: + """List annotation jobs through the administration endpoint with cursor pagination.""" + return rfapi.list_annotation_jobs_admin( + self.__api_key, + self.__workspace, + self.__project_name, + limit=limit, + after=after, + show_empty=show_empty, + ) + def get_annotation_job(self, job_id: str) -> Dict: """Get information for a specific annotation job. @@ -903,63 +965,195 @@ def get_annotation_job(self, job_id: str) -> Dict: Returns: Dict: A dictionary containing the job details. """ - from roboflow.adapters import rfapi - return rfapi.get_annotation_job(self.__api_key, self.__workspace, self.__project_name, job_id) + def get_annotation_job_admin(self, job_id: str) -> Dict: + """Get one annotation job through the administration endpoint.""" + return rfapi.get_annotation_job_admin(self.__api_key, self.__workspace, self.__project_name, job_id) + + def get_annotation_job_images(self, job_id: str, limit: int = 50, after: Optional[str] = None) -> Dict: + """List image IDs assigned to an annotation job with cursor pagination.""" + return rfapi.list_annotation_job_images( + self.__api_key, + self.__workspace, + self.__project_name, + job_id, + limit=limit, + after=after, + ) + def create_annotation_job( - self, name: str, batch_id: str, num_images: int, labeler_email: str, reviewer_email: str + self, + name: Optional[str] = None, + batch_id: Optional[str] = None, + num_images: Optional[int] = None, + labeler_email: Optional[str] = None, + reviewer_email: Optional[str] = None, + instructions: Optional[str] = None, ) -> Dict: + """Create a job and move images from a batch into it. + + ``name`` and ``num_images`` are optional. Their position is retained for + compatibility with older positional callers. """ - Create a new annotation job in the project. + if not batch_id or not labeler_email or not reviewer_email: + raise ValueError("batch_id, labeler_email, and reviewer_email are required") + try: + return rfapi.create_annotation_job_from_batch( + self.__api_key, + self.__workspace, + self.__project_name, + batch_id=batch_id, + labeler_email=labeler_email, + reviewer_email=reviewer_email, + name=name, + num_images=num_images, + instructions=instructions, + ) + except rfapi.RoboflowError as exc: + # This public method historically raised RuntimeError for API failures. + raise RuntimeError(str(exc)) from exc - Args: - name (str): The name of the annotation job - batch_id (str): The ID of the batch that contains the images to annotate - num_images (int): The number of images to include in the job - labeler_email (str): The email of the user who will label the images - reviewer_email (str): The email of the user who will review the annotations + def create_annotation_job_admin( + self, + name: Optional[str] = None, + batch_id: Optional[str] = None, + num_images: Optional[int] = None, + labeler_email: Optional[str] = None, + reviewer_email: Optional[str] = None, + instructions: Optional[str] = None, + ) -> Dict: + """Create a job through the administration endpoint.""" + if not batch_id or not labeler_email or not reviewer_email: + raise ValueError("batch_id, labeler_email, and reviewer_email are required") + return rfapi.create_annotation_job_admin( + self.__api_key, + self.__workspace, + self.__project_name, + batch_id=batch_id, + labeler_email=labeler_email, + reviewer_email=reviewer_email, + name=name, + num_images=num_images, + instructions=instructions, + ) - Returns: - Dict: A dictionary containing the created job details + def reassign_annotation_job_images( + self, + image_ids: List[str], + labeler_email: str, + reviewer_email: Optional[str] = None, + instructions: Optional[str] = None, + name: Optional[str] = None, + ) -> Dict: + """Create a job by removing selected images from their prior assignment.""" + return rfapi.reassign_annotation_job_images( + self.__api_key, + self.__workspace, + self.__project_name, + image_ids=image_ids, + labeler_email=labeler_email, + reviewer_email=reviewer_email, + instructions=instructions, + name=name, + ) - Example: - >>> import roboflow + def add_annotation_job_images(self, job_id: str, image_ids: List[str]) -> Dict: + """Move selected images into an existing annotation job.""" + return rfapi.add_images_to_annotation_job( + self.__api_key, + self.__workspace, + self.__project_name, + job_id, + image_ids=image_ids, + ) - >>> rf = roboflow.Roboflow(api_key="YOUR_API_KEY") + def update_annotation_job( + self, + job_id: str, + *, + labeler_email: Optional[str] = None, + reviewer_email: Optional[str] = None, + instructions: Optional[str] = None, + ) -> Dict: + """Update exactly one of a job's labeler, reviewer, or instructions.""" + return rfapi.update_annotation_job( + self.__api_key, + self.__workspace, + self.__project_name, + job_id, + labeler_email=labeler_email, + reviewer_email=reviewer_email, + instructions=instructions, + ) - >>> project = rf.workspace().project("PROJECT_ID") + def submit_annotation_job_for_review(self, job_id: str) -> Dict: + """Advance a labeling job into review.""" + return rfapi.submit_annotation_job_for_review(self.__api_key, self.__workspace, self.__project_name, job_id) + + def return_annotation_job_for_edits(self, job_id: str, new_labeler_email: Optional[str] = None) -> Dict: + """Move a review job back to labeling, optionally with a new labeler.""" + return rfapi.return_annotation_job_for_edits( + self.__api_key, + self.__workspace, + self.__project_name, + job_id, + new_labeler_email=new_labeler_email, + ) - >>> job = project.create_annotation_job( - ... name="Job created by API", - ... batch_id="batch123", - ... num_images=10, - ... labeler_email="user@example.com", - ... reviewer_email="reviewer@example.com" - ... ) - """ - url = f"{API_URL}/{self.__workspace}/{self.__project_name}/jobs?api_key={self.__api_key}" - - payload = { - "name": name, - "batch": batch_id, - "num_images": num_images, - "labelerEmail": labeler_email, - "reviewerEmail": reviewer_email, - } + def review_annotation_job_image(self, job_id: str, image_id: str, status: str) -> Dict: + """Set the review status for one image in a job.""" + return rfapi.review_annotation_job_image( + self.__api_key, + self.__workspace, + self.__project_name, + job_id, + image_id, + status=status, + ) - response = requests.post(url, headers={"Content-Type": "application/json"}, json=payload) + def review_annotation_job_images(self, job_id: str, status: str, current_status: str) -> Dict: + """Set a status for every job image matching the supplied current status.""" + return rfapi.review_annotation_job_images( + self.__api_key, + self.__workspace, + self.__project_name, + job_id, + status=status, + current_status=current_status, + ) - 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 create annotation job: {response.text}") + def accept_annotation_job_images( + self, + job_id: str, + split_method: str, + statuses_to_include: List[str], + train_count: int, + valid_count: int, + test_count: int, + image_ids: Optional[List[str]] = None, + ) -> Dict: + """Accept selected job images into Dataset and assign their splits.""" + return rfapi.accept_annotation_job_images( + self.__api_key, + self.__workspace, + self.__project_name, + job_id, + split_method=split_method, + statuses_to_include=statuses_to_include, + train_count=train_count, + valid_count=valid_count, + test_count=test_count, + image_ids=image_ids, + ) - return response.json() + def move_annotation_job_to_unassigned(self, job_id: str) -> Dict: + """Remove a job while retaining its images in an unassigned batch.""" + return rfapi.move_annotation_job_to_unassigned(self.__api_key, self.__workspace, self.__project_name, job_id) + + def delete_annotation_job_annotations(self, job_id: str) -> Dict: + """Delete project annotations from every image assigned to a job.""" + return rfapi.delete_annotation_job_annotations(self.__api_key, self.__workspace, self.__project_name, job_id) def get_batches(self) -> Dict: """ diff --git a/tests/adapters/test_annotation_administration.py b/tests/adapters/test_annotation_administration.py new file mode 100644 index 00000000..42460843 --- /dev/null +++ b/tests/adapters/test_annotation_administration.py @@ -0,0 +1,219 @@ +"""HTTP contract tests for annotation administration adapters.""" + +import unittest +from unittest.mock import MagicMock, patch + +from roboflow.adapters import rfapi + + +def _response(payload=None, status_code=200): + return MagicMock(status_code=status_code, text="error", json=lambda: payload or {"success": True}) + + +class TestAnnotationBatchAdministration(unittest.TestCase): + @patch("roboflow.adapters.rfapi.requests.get") + def test_read_contracts(self, mock_get): + mock_get.return_value = _response() + + rfapi.list_annotation_batches("key", "ws", "proj", limit=25, after="cursor", show_empty=True) + self.assertTrue(mock_get.call_args.args[0].endswith("/ws/proj/annotation-batches")) + self.assertEqual( + mock_get.call_args.kwargs["params"], + {"api_key": "key", "limit": 25, "after": "cursor", "showEmpty": "true"}, + ) + + rfapi.get_annotation_batch("key", "ws", "proj", "batch-1") + self.assertTrue(mock_get.call_args.args[0].endswith("/annotation-batches/batch-1")) + + rfapi.list_annotation_batch_images("key", "ws", "proj", "batch-1", limit=10, after="next") + self.assertTrue(mock_get.call_args.args[0].endswith("/annotation-batches/batch-1/images")) + self.assertEqual( + mock_get.call_args.kwargs["params"], + {"api_key": "key", "limit": 10, "after": "next"}, + ) + + @patch("roboflow.adapters.rfapi.requests.post") + def test_create_and_merge_contracts(self, mock_post): + mock_post.return_value = _response() + + rfapi.create_annotation_batch( + "key", + "ws", + "proj", + source_batch_id="source", + image_ids=["image-1", "image-2"], + name="Round two", + ) + self.assertTrue(mock_post.call_args.args[0].endswith("/ws/proj/annotation-batches")) + self.assertEqual( + mock_post.call_args.kwargs["json"], + {"sourceBatchId": "source", "imageIds": ["image-1", "image-2"], "name": "Round two"}, + ) + + rfapi.merge_annotation_batches( + "key", "ws", "proj", source_batch_ids=["source-1", "source-2"], target_batch_id="target" + ) + self.assertTrue(mock_post.call_args.args[0].endswith("/annotation-batches/merge")) + self.assertEqual( + mock_post.call_args.kwargs["json"], + {"sourceBatchIds": ["source-1", "source-2"], "targetBatchId": "target"}, + ) + + @patch("roboflow.adapters.rfapi.requests.delete") + def test_delete_contract(self, mock_delete): + mock_delete.return_value = _response() + rfapi.delete_annotation_batch("key", "ws", "proj", "batch-1", permanent=True) + self.assertTrue(mock_delete.call_args.args[0].endswith("/annotation-batches/batch-1")) + self.assertEqual(mock_delete.call_args.kwargs["params"], {"api_key": "key", "permanent": "true"}) + + +class TestAnnotationJobAdministration(unittest.TestCase): + @patch("roboflow.adapters.rfapi.requests.get") + def test_read_contracts(self, mock_get): + mock_get.return_value = _response() + + rfapi.list_annotation_jobs_admin("key", "ws", "proj", limit=20, after="cursor", show_empty=True) + self.assertTrue(mock_get.call_args.args[0].endswith("/ws/proj/annotation-jobs")) + self.assertEqual( + mock_get.call_args.kwargs["params"], + {"api_key": "key", "limit": 20, "after": "cursor", "showEmpty": "true"}, + ) + + rfapi.get_annotation_job_admin("key", "ws", "proj", "job-1") + self.assertTrue(mock_get.call_args.args[0].endswith("/annotation-jobs/job-1")) + + rfapi.list_annotation_job_images("key", "ws", "proj", "job-1", limit=5, after="next") + self.assertTrue(mock_get.call_args.args[0].endswith("/annotation-jobs/job-1/images")) + self.assertEqual( + mock_get.call_args.kwargs["params"], + {"api_key": "key", "limit": 5, "after": "next"}, + ) + + @patch("roboflow.adapters.rfapi.requests.post") + def test_create_reassign_and_add_contracts(self, mock_post): + mock_post.return_value = _response() + + rfapi.create_annotation_job_admin( + "key", + "ws", + "proj", + batch_id="batch-1", + labeler_email="labeler@example.com", + reviewer_email="reviewer@example.com", + name="Round one", + num_images=12, + instructions="Follow the guide", + ) + self.assertTrue(mock_post.call_args.args[0].endswith("/ws/proj/annotation-jobs")) + self.assertEqual( + mock_post.call_args.kwargs["json"], + { + "batchId": "batch-1", + "labelerEmail": "labeler@example.com", + "reviewerEmail": "reviewer@example.com", + "name": "Round one", + "numImages": 12, + "instructions": "Follow the guide", + }, + ) + + rfapi.reassign_annotation_job_images( + "key", + "ws", + "proj", + image_ids=["image-1"], + labeler_email="labeler@example.com", + reviewer_email="reviewer@example.com", + name="Reassigned", + ) + self.assertTrue(mock_post.call_args.args[0].endswith("/annotation-jobs/reassign-images")) + self.assertEqual(mock_post.call_args.kwargs["json"]["imageIds"], ["image-1"]) + self.assertNotIn("instructions", mock_post.call_args.kwargs["json"]) + + rfapi.add_images_to_annotation_job("key", "ws", "proj", "job-1", image_ids=["image-2"]) + self.assertTrue(mock_post.call_args.args[0].endswith("/annotation-jobs/job-1/images")) + self.assertEqual(mock_post.call_args.kwargs["json"], {"imageIds": ["image-2"]}) + + @patch("roboflow.adapters.rfapi.requests.patch") + def test_update_contract_and_validation(self, mock_patch): + mock_patch.return_value = _response() + rfapi.update_annotation_job("key", "ws", "proj", "job-1", reviewer_email="new@example.com") + self.assertEqual(mock_patch.call_args.kwargs["json"], {"reviewerEmail": "new@example.com"}) + + with self.assertRaisesRegex(ValueError, "exactly one"): + rfapi.update_annotation_job("key", "ws", "proj", "job-1") + with self.assertRaisesRegex(ValueError, "exactly one"): + rfapi.update_annotation_job("key", "ws", "proj", "job-1", labeler_email="a@example.com", instructions="new") + + @patch("roboflow.adapters.rfapi.requests.post") + def test_review_transition_contracts(self, mock_post): + mock_post.return_value = _response() + + rfapi.submit_annotation_job_for_review("key", "ws", "proj", "job-1") + self.assertTrue(mock_post.call_args.args[0].endswith("/annotation-jobs/job-1/submit-review")) + self.assertEqual(mock_post.call_args.kwargs["json"], {}) + + rfapi.return_annotation_job_for_edits("key", "ws", "proj", "job-1", new_labeler_email="new@example.com") + self.assertTrue(mock_post.call_args.args[0].endswith("/annotation-jobs/job-1/return-edits")) + self.assertEqual(mock_post.call_args.kwargs["json"], {"newLabelerEmail": "new@example.com"}) + + rfapi.review_annotation_job_image("key", "ws", "proj", "job-1", "image-1", status="approved") + self.assertTrue(mock_post.call_args.args[0].endswith("/annotation-jobs/job-1/images/image-1/status")) + self.assertEqual(mock_post.call_args.kwargs["json"], {"status": "approved"}) + + rfapi.review_annotation_job_images("key", "ws", "proj", "job-1", status="rejected", current_status="annotated") + self.assertTrue(mock_post.call_args.args[0].endswith("/annotation-jobs/job-1/images/status")) + self.assertEqual( + mock_post.call_args.kwargs["json"], + {"status": "rejected", "currentStatus": "annotated"}, + ) + + @patch("roboflow.adapters.rfapi.requests.post") + def test_accept_and_move_contracts(self, mock_post): + mock_post.return_value = _response() + + rfapi.accept_annotation_job_images( + "key", + "ws", + "proj", + "job-1", + split_method="split", + statuses_to_include=["approved", "annotated"], + train_count=8, + valid_count=1, + test_count=1, + image_ids=["image-1"], + ) + self.assertTrue(mock_post.call_args.args[0].endswith("/annotation-jobs/job-1/accept")) + self.assertEqual( + mock_post.call_args.kwargs["json"], + { + "splitMethod": "split", + "statusesToInclude": ["approved", "annotated"], + "trainCount": 8, + "validCount": 1, + "testCount": 1, + "imageIds": ["image-1"], + }, + ) + + rfapi.move_annotation_job_to_unassigned("key", "ws", "proj", "job-1") + self.assertTrue(mock_post.call_args.args[0].endswith("/annotation-jobs/job-1/move-to-unassigned")) + self.assertEqual(mock_post.call_args.kwargs["json"], {}) + + @patch("roboflow.adapters.rfapi.requests.delete") + def test_delete_annotations_contract(self, mock_delete): + mock_delete.return_value = _response() + rfapi.delete_annotation_job_annotations("key", "ws", "proj", "job-1") + self.assertTrue(mock_delete.call_args.args[0].endswith("/annotation-jobs/job-1/annotations")) + + @patch("roboflow.adapters.rfapi.requests.get") + def test_http_error_preserves_status_code(self, mock_get): + mock_get.return_value = _response(status_code=403) + with self.assertRaises(rfapi.RoboflowError) as context: + rfapi.list_annotation_jobs("key", "ws", "proj") + self.assertEqual(context.exception.status_code, 403) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/adapters/test_rfapi_phase2.py b/tests/adapters/test_rfapi_phase2.py index dc03e800..a5b43f41 100644 --- a/tests/adapters/test_rfapi_phase2.py +++ b/tests/adapters/test_rfapi_phase2.py @@ -55,6 +55,7 @@ def test_success(self, mock_get): result = list_annotation_jobs("key", "ws", "proj") self.assertEqual(result, {"jobs": []}) self.assertIn("/ws/proj/jobs", mock_get.call_args[0][0]) + self.assertEqual(mock_get.call_args.kwargs["params"], {"api_key": "key"}) @patch("roboflow.adapters.rfapi.requests.get") def test_error(self, mock_get): diff --git a/tests/cli/test_annotation_handler.py b/tests/cli/test_annotation_handler.py index 8671ff7b..e2a9676a 100644 --- a/tests/cli/test_annotation_handler.py +++ b/tests/cli/test_annotation_handler.py @@ -5,7 +5,7 @@ import sys import types import unittest -from unittest.mock import MagicMock, patch +from unittest.mock import patch from typer.testing import CliRunner @@ -37,6 +37,32 @@ def test_annotation_job_create(self): result = runner.invoke(app, ["annotation", "job", "create", "--help"]) self.assertEqual(result.exit_code, 0) + def test_full_annotation_administration_surface(self): + commands = { + "batch": ["admin-list", "admin-get", "images", "create", "merge", "delete"], + "job": [ + "admin-list", + "admin-get", + "admin-create", + "images", + "reassign-images", + "add-images", + "update", + "submit-review", + "return-edits", + "review-image", + "review-images", + "accept", + "move-to-unassigned", + "delete-annotations", + ], + } + for group, names in commands.items(): + for name in names: + with self.subTest(command=f"{group} {name}"): + result = runner.invoke(app, ["annotation", group, name, "--help"]) + self.assertEqual(result.exit_code, 0, result.output) + class TestAnnotationStub(unittest.TestCase): """Verify stub handlers print not-yet-implemented.""" @@ -161,12 +187,10 @@ def test_text_output(self, _resolve, mock_api): class TestJobCreate(unittest.TestCase): """annotation job create""" - @patch("roboflow.Roboflow") + @patch("roboflow.adapters.rfapi.create_annotation_job_from_batch") @patch(_RESOLVE, return_value=("key", "ws", "proj")) - def test_text_output(self, _resolve, mock_rf_cls): - mock_project = MagicMock() - mock_project.create_annotation_job.return_value = {"id": "42", "name": "new-job"} - mock_rf_cls.return_value.workspace.return_value.project.return_value = mock_project + def test_text_output(self, _resolve, mock_api): + mock_api.return_value = {"id": "42", "name": "new-job"} result = runner.invoke( app, @@ -189,20 +213,22 @@ def test_text_output(self, _resolve, mock_rf_cls): ], ) self.assertIn("new-job", result.output) - mock_project.create_annotation_job.assert_called_once_with( - name="new-job", + mock_api.assert_called_once_with( + "key", + "ws", + "proj", batch_id="b1", - num_images=5, labeler_email="a@b.com", reviewer_email="c@d.com", + name="new-job", + num_images=5, + instructions=None, ) - @patch("roboflow.Roboflow") + @patch("roboflow.adapters.rfapi.create_annotation_job_from_batch") @patch(_RESOLVE, return_value=("key", "ws", "proj")) - def test_json_output(self, _resolve, mock_rf_cls): - mock_project = MagicMock() - mock_project.create_annotation_job.return_value = {"id": "42", "name": "new-job"} - mock_rf_cls.return_value.workspace.return_value.project.return_value = mock_project + def test_json_output(self, _resolve, mock_api): + mock_api.return_value = {"id": "42", "name": "new-job"} result = runner.invoke( app, @@ -251,5 +277,214 @@ def test_create_requires_all_flags(self): self.assertNotEqual(result.exit_code, 0) +class TestAnnotationAdministrationCommands(unittest.TestCase): + @patch("roboflow.adapters.rfapi.list_annotation_jobs_admin") + @patch(_RESOLVE, return_value=("key", "ws", "proj")) + def test_admin_list_preserves_pagination_response(self, _resolve, mock_api): + mock_api.return_value = {"jobs": [{"id": "job-1"}], "continuationToken": "next"} + result = runner.invoke( + app, + [ + "--json", + "annotation", + "job", + "admin-list", + "-p", + "ws/proj", + "--limit", + "10", + "--after", + "cursor", + "--show-empty", + ], + ) + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual(json.loads(result.output)["continuationToken"], "next") + mock_api.assert_called_once_with("key", "ws", "proj", limit=10, after="cursor", show_empty=True) + + @patch("roboflow.adapters.rfapi.get_annotation_job_admin") + @patch(_RESOLVE, return_value=("key", "ws", "proj")) + def test_admin_get_uses_administration_adapter(self, _resolve, mock_api): + mock_api.return_value = {"id": "job-1"} + result = runner.invoke(app, ["--json", "annotation", "job", "admin-get", "job-1", "-p", "ws/proj"]) + self.assertEqual(result.exit_code, 0, result.output) + mock_api.assert_called_once_with("key", "ws", "proj", "job-1") + + @patch("roboflow.adapters.rfapi.create_annotation_job_admin") + @patch(_RESOLVE, return_value=("key", "ws", "proj")) + def test_admin_create_uses_administration_adapter(self, _resolve, mock_api): + mock_api.return_value = {"id": "job-1"} + result = runner.invoke( + app, + [ + "--json", + "annotation", + "job", + "admin-create", + "-p", + "ws/proj", + "--batch", + "batch-1", + "--labeler", + "labeler@example.com", + "--reviewer", + "reviewer@example.com", + ], + ) + self.assertEqual(result.exit_code, 0, result.output) + mock_api.assert_called_once_with( + "key", + "ws", + "proj", + batch_id="batch-1", + labeler_email="labeler@example.com", + reviewer_email="reviewer@example.com", + name=None, + num_images=None, + instructions=None, + ) + + @patch("roboflow.adapters.rfapi.create_annotation_batch") + @patch(_RESOLVE, return_value=("key", "ws", "proj")) + def test_batch_create_accepts_repeated_image_ids(self, _resolve, mock_api): + mock_api.return_value = {"id": "batch-2"} + result = runner.invoke( + app, + [ + "--json", + "annotation", + "batch", + "create", + "-p", + "ws/proj", + "--source-batch-id", + "batch-1", + "--image-id", + "image-1", + "--image-id", + "image-2", + "--name", + "Round two", + ], + ) + self.assertEqual(result.exit_code, 0, result.output) + mock_api.assert_called_once_with( + "key", + "ws", + "proj", + source_batch_id="batch-1", + image_ids=["image-1", "image-2"], + name="Round two", + ) + + @patch("roboflow.adapters.rfapi.merge_annotation_batches") + @patch(_RESOLVE, return_value=("key", "ws", "proj")) + def test_batch_merge_requires_yes_non_interactively(self, _resolve, mock_api): + command = [ + "annotation", + "batch", + "merge", + "-p", + "ws/proj", + "--source-batch-id", + "source", + "--target-batch-id", + "target", + ] + result = runner.invoke(app, command) + self.assertEqual(result.exit_code, 1) + mock_api.assert_not_called() + + mock_api.return_value = {"success": True} + result = runner.invoke(app, [*command, "--yes"]) + self.assertEqual(result.exit_code, 0, result.output) + mock_api.assert_called_once_with("key", "ws", "proj", source_batch_ids=["source"], target_batch_id="target") + + @patch("roboflow.adapters.rfapi.add_images_to_annotation_job") + @patch(_RESOLVE, return_value=("key", "ws", "proj")) + def test_job_add_images_requires_yes_non_interactively(self, _resolve, mock_api): + command = [ + "annotation", + "job", + "add-images", + "job-1", + "-p", + "ws/proj", + "--image-id", + "image-1", + ] + result = runner.invoke(app, command) + self.assertEqual(result.exit_code, 1) + mock_api.assert_not_called() + + mock_api.return_value = {"movedImageCount": 1} + result = runner.invoke(app, [*command, "--yes"]) + self.assertEqual(result.exit_code, 0, result.output) + mock_api.assert_called_once_with("key", "ws", "proj", "job-1", image_ids=["image-1"]) + + @patch("roboflow.adapters.rfapi.accept_annotation_job_images") + @patch(_RESOLVE, return_value=("key", "ws", "proj")) + def test_job_accept_maps_lists_and_split_counts(self, _resolve, mock_api): + mock_api.return_value = {"success": True, "numImagesAdded": 1} + result = runner.invoke( + app, + [ + "--json", + "annotation", + "job", + "accept", + "job-1", + "-p", + "ws/proj", + "--split-method", + "split", + "--status", + "approved", + "--status", + "annotated", + "--train-count", + "1", + "--valid-count", + "0", + "--test-count", + "0", + "--image-id", + "image-1", + "--yes", + ], + ) + self.assertEqual(result.exit_code, 0, result.output) + mock_api.assert_called_once_with( + "key", + "ws", + "proj", + "job-1", + split_method="split", + statuses_to_include=["approved", "annotated"], + train_count=1, + valid_count=0, + test_count=0, + image_ids=["image-1"], + ) + + @patch("roboflow.adapters.rfapi.update_annotation_job", side_effect=ValueError("Provide exactly one field")) + @patch(_RESOLVE, return_value=("key", "ws", "proj")) + def test_job_update_surfaces_validation_error(self, _resolve, _mock_api): + result = runner.invoke(app, ["annotation", "job", "update", "job-1", "-p", "ws/proj"]) + self.assertEqual(result.exit_code, 1) + self.assertIn("exactly one", result.output) + + @patch("roboflow.adapters.rfapi.delete_annotation_job_annotations") + @patch(_RESOLVE, return_value=("key", "ws", "proj")) + def test_delete_annotations_executes_with_yes(self, _resolve, mock_api): + mock_api.return_value = {"success": True} + result = runner.invoke( + app, + ["annotation", "job", "delete-annotations", "job-1", "-p", "ws/proj", "--yes"], + ) + self.assertEqual(result.exit_code, 0, result.output) + mock_api.assert_called_once_with("key", "ws", "proj", "job-1") + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_project_annotation_administration.py b/tests/test_project_annotation_administration.py new file mode 100644 index 00000000..725573a4 --- /dev/null +++ b/tests/test_project_annotation_administration.py @@ -0,0 +1,118 @@ +"""Public Project wrapper coverage for annotation administration.""" + +from unittest.mock import patch + +from tests import RoboflowTest + + +class TestProjectAnnotationAdministration(RoboflowTest): + def test_batch_wrappers_delegate_to_rfapi(self): + cases = [ + ( + "get_annotation_batches", + (), + {"limit": 10, "after": "cursor", "show_empty": True}, + "list_annotation_batches", + ), + ("get_annotation_batch", ("batch-1",), {}, "get_annotation_batch"), + ( + "get_annotation_batch_images", + ("batch-1",), + {"limit": 5, "after": "next"}, + "list_annotation_batch_images", + ), + ( + "create_annotation_batch", + ("source", ["image-1"]), + {"name": "New batch"}, + "create_annotation_batch", + ), + ("merge_annotation_batches", (["source"], "target"), {}, "merge_annotation_batches"), + ("delete_annotation_batch", ("batch-1",), {"permanent": True}, "delete_annotation_batch"), + ] + + for method, args, kwargs, adapter in cases: + with ( + self.subTest(method=method), + patch(f"roboflow.adapters.rfapi.{adapter}", return_value={"ok": True}) as mock, + ): + self.assertEqual(getattr(self.project, method)(*args, **kwargs), {"ok": True}) + mock.assert_called_once() + + def test_job_wrappers_delegate_to_rfapi(self): + cases = [ + ("get_annotation_jobs", (), {}, "list_annotation_jobs"), + ( + "get_annotation_jobs_admin", + (), + {"limit": 10, "after": "cursor", "show_empty": True}, + "list_annotation_jobs_admin", + ), + ("get_annotation_job", ("job-1",), {}, "get_annotation_job"), + ("get_annotation_job_admin", ("job-1",), {}, "get_annotation_job_admin"), + ("get_annotation_job_images", ("job-1",), {"limit": 5, "after": "next"}, "list_annotation_job_images"), + ( + "create_annotation_job", + (), + { + "batch_id": "batch-1", + "labeler_email": "labeler@example.com", + "reviewer_email": "reviewer@example.com", + "instructions": "Guide", + }, + "create_annotation_job_from_batch", + ), + ( + "create_annotation_job_admin", + (), + { + "batch_id": "batch-1", + "labeler_email": "labeler@example.com", + "reviewer_email": "reviewer@example.com", + "instructions": "Guide", + }, + "create_annotation_job_admin", + ), + ( + "reassign_annotation_job_images", + (["image-1"], "labeler@example.com"), + {"reviewer_email": "reviewer@example.com"}, + "reassign_annotation_job_images", + ), + ("add_annotation_job_images", ("job-1", ["image-1"]), {}, "add_images_to_annotation_job"), + ("update_annotation_job", ("job-1",), {"instructions": "New guide"}, "update_annotation_job"), + ("submit_annotation_job_for_review", ("job-1",), {}, "submit_annotation_job_for_review"), + ( + "return_annotation_job_for_edits", + ("job-1",), + {"new_labeler_email": "new@example.com"}, + "return_annotation_job_for_edits", + ), + ("review_annotation_job_image", ("job-1", "image-1", "approved"), {}, "review_annotation_job_image"), + ( + "review_annotation_job_images", + ("job-1", "approved", "annotated"), + {}, + "review_annotation_job_images", + ), + ( + "accept_annotation_job_images", + ("job-1", "split", ["approved"], 8, 1, 1), + {"image_ids": ["image-1"]}, + "accept_annotation_job_images", + ), + ("move_annotation_job_to_unassigned", ("job-1",), {}, "move_annotation_job_to_unassigned"), + ("delete_annotation_job_annotations", ("job-1",), {}, "delete_annotation_job_annotations"), + ] + + for method, args, kwargs, adapter in cases: + with ( + self.subTest(method=method), + patch(f"roboflow.adapters.rfapi.{adapter}", return_value={"ok": True}) as mock, + ): + self.assertEqual(getattr(self.project, method)(*args, **kwargs), {"ok": True}) + mock.assert_called_once() + + def test_create_job_validates_required_assignments(self): + with self.assertRaisesRegex(ValueError, "required"): + self.project.create_annotation_job(batch_id="batch-1")