Honor annotation_overwrite in per-image upload_dataset (REQ-182)

roboflow/cli/handlers/_aliases.py

@@ -76,6 +76,14 @@ def upload_alias(
         metadata: Annotated[Optional[str], typer.Option("-M", "--metadata", help="JSON metadata")] = None,
         concurrency: Annotated[int, typer.Option("-c", "--concurrency", help="Concurrency")] = 10,
         is_prediction: Annotated[bool, typer.Option("--is-prediction", help="Mark as prediction")] = False,
+        annotation_overwrite: Annotated[
+            Optional[bool],
+            typer.Option(
+                "--annotation-overwrite/--no-annotation-overwrite",
+                help="Overwrite existing annotations on duplicate images "
+                "(default: off, except classification projects where the API requires it on)",
+            ),
+        ] = None,
     ) -> None:
         """Upload images to a project (alias for 'image upload')."""
         from roboflow.cli.handlers.image import _handle_upload
@@ -93,6 +101,7 @@ def upload_alias(
             metadata=metadata,
             concurrency=concurrency,
             is_prediction=is_prediction,
+            annotation_overwrite=annotation_overwrite,
         )
         _handle_upload(args)
 

roboflow/cli/handlers/image.py

@@ -44,7 +44,7 @@ def upload_image(
         Optional[bool],
         typer.Option(
             "--annotation-overwrite/--no-annotation-overwrite",
-            help="Zip flow: overwrite existing annotations on duplicate images "
+            help="Overwrite existing annotations on duplicate images "
             "(default: off, except classification projects where the API requires it on)",
         ),
     ] = None,

roboflow/core/workspace.py

@@ -540,7 +540,7 @@ def upload_dataset(
             num_retries (int, optional): number of times to retry uploading an image if the upload fails. Defaults to 0.
             is_prediction (bool, optional): whether the annotations provided in the dataset are predictions and not ground truth. Defaults to False.
             use_zip_upload (bool, optional): opt-in to the zip flow for a directory input (the SDK zips it client-side). Ignored when dataset_path is already a `.zip`.
-            annotation_overwrite (bool, optional): zip flow only — overwrite existing annotations on duplicate images. Defaults to False, except classification projects where it defaults to True (the API requires it).
+            annotation_overwrite (bool, optional): overwrite existing annotations on duplicate images. Defaults to False, except classification projects where it defaults to True (the API requires it).
             tags (list[str], optional): zip flow only — tags to apply to the uploaded batch.
             split (str, optional): dataset split for the uploaded batch. In per-image directory
                 uploads, this overrides inferred splits for every image.
@@ -569,6 +569,9 @@ def upload_dataset(
                 "Call upload_dataset without use_zip_upload for prediction uploads."
             )
 
+        if annotation_overwrite is None:
+            annotation_overwrite = project.type == "classification"
+
         if use_zip_flow:
             project_slug = project.id.rsplit("/")[1]
             temp_zip = None
@@ -579,9 +582,6 @@ def upload_dataset(
                     zip_path = temp_zip = _zip_directory(dataset_path)
                     print(f"Zipped {dataset_path} -> {zip_path}")
 
-                if annotation_overwrite is None:
-                    annotation_overwrite = project.type == "classification"
-
                 init = rfapi.init_zip_upload(
                     self.__api_key,
                     self.url,
@@ -635,6 +635,11 @@ def _log_img_upload(
             if annotation:
                 if annotation.get("success"):
                     msg += f" / annotations = OK {annotation_time_str}"
+                elif annotation.get("warn") == "already annotated":
+                    msg += (
+                        " / annotations = kept existing (pass --annotation-overwrite / annotation_overwrite=True"
+                        f" to replace) {annotation_time_str}"
+                    )
                 elif annotation.get("warn"):
                     msg += f" / annotations = WARN: {annotation['warn']} {annotation_time_str}"
                 else:
@@ -687,6 +692,7 @@ def _save_annotation(image_id, imagedesc):
                 job_name=batch_name,
                 num_retry_uploads=num_retries,
                 is_prediction=is_prediction,
+                annotation_overwrite=annotation_overwrite,
             )
 
             return annotation, upload_time

tests/cli/test_image_handler.py

@@ -348,19 +348,20 @@ def test_zip_upload_flag_defaults_false(self, mock_rf_cls):
     @patch("roboflow.cli.handlers.image._handle_upload")
     def test_annotation_overwrite_flag_three_states(self, mock_handle_upload):
         with tempfile.TemporaryDirectory() as tmpdir:
-            for extra_argv, expected in [
-                ([], None),
-                (["--annotation-overwrite"], True),
-                (["--no-annotation-overwrite"], False),
-            ]:
-                mock_handle_upload.reset_mock()
-                result = runner.invoke(
-                    app,
-                    ["--workspace", "ws", "--api-key", "k", "image", "upload", tmpdir, "-p", "proj"] + extra_argv,
-                )
-                self.assertEqual(result.exit_code, 0)
-                args = mock_handle_upload.call_args.args[0]
-                self.assertEqual(args.annotation_overwrite, expected)
+            for command in (["image", "upload"], ["upload"]):
+                for extra_argv, expected in [
+                    ([], None),
+                    (["--annotation-overwrite"], True),
+                    (["--no-annotation-overwrite"], False),
+                ]:
+                    mock_handle_upload.reset_mock()
+                    result = runner.invoke(
+                        app,
+                        ["--workspace", "ws", "--api-key", "k", *command, tmpdir, "-p", "proj"] + extra_argv,
+                    )
+                    self.assertEqual(result.exit_code, 0, result.output)
+                    args = mock_handle_upload.call_args.args[0]
+                    self.assertEqual(args.annotation_overwrite, expected)
 
     @patch("roboflow.Roboflow")
     def test_annotation_overwrite_forwarded_to_upload_dataset(self, mock_rf_cls):

tests/test_project.py

@@ -1218,3 +1218,70 @@ def test_poll_loop_timeout(self):
 
             if _os.path.exists(zip_path):
                 _os.unlink(zip_path)
+
+
+class TestPerImageAnnotationOverwrite(RoboflowTest):
+    """upload_dataset on a directory (per-image flow) forwards annotation_overwrite to the annotate POST."""
+
+    def _upload(self, image, annotate_status=200, annotate_json=None, **kwargs):
+        import re
+
+        dataset = {
+            "location": "/test/location/",
+            "images": [
+                {
+                    "file": "img.jpg",
+                    "split": "train",
+                    "annotationfile": {"name": "annotation.coco.json", "rawText": "{}"},
+                }
+            ],
+        }
+        responses.add(
+            responses.POST,
+            re.compile(rf"{API_URL}/dataset/{PROJECT_NAME}/annotate/img-id\?.*"),
+            json=annotate_json or {"success": True},
+            status=annotate_status,
+        )
+        with (
+            patch("roboflow.util.folderparser.parsefolder", return_value=dataset),
+            patch("roboflow.core.project.Project.upload_image", return_value=(image, 0.1, 0)),
+            patch("roboflow.core.workspace.Workspace._get_or_create_project", return_value=(self.project, False)),
+            patch("builtins.print") as mock_print,
+        ):
+            self.workspace.upload_dataset(dataset_path="/test/dataset", project_name=PROJECT_NAME, **kwargs)
+        annotate_calls = [c for c in responses.calls if "/annotate/" in c.request.url]
+        logs = " ".join(str(c.args[0]) for c in mock_print.call_args_list if c.args)
+        return annotate_calls, logs
+
+    @responses.activate
+    def test_fresh_image_annotates_without_overwrite(self):
+        calls, logs = self._upload({"id": "img-id", "success": True})
+        self.assertEqual(len(calls), 1)
+        self.assertNotIn("overwrite=true", calls[0].request.url)
+        self.assertIn("annotations = OK", logs)
+
+    @responses.activate
+    def test_duplicate_without_overwrite_keeps_existing_annotations(self):
+        calls, logs = self._upload(
+            {"id": "img-id", "duplicate": True},
+            annotate_status=409,
+            annotate_json={"error": {"message": "Image was already annotated. To overwrite, pass overwrite=true."}},
+        )
+        self.assertEqual(len(calls), 1)
+        self.assertNotIn("overwrite=true", calls[0].request.url)
+        self.assertIn(
+            "annotations = kept existing (pass --annotation-overwrite / annotation_overwrite=True to replace)", logs
+        )
+
+    @responses.activate
+    def test_duplicate_with_overwrite_true_sends_overwrite(self):
+        calls, _ = self._upload({"id": "img-id", "duplicate": True}, annotation_overwrite=True)
+        self.assertEqual(len(calls), 1)
+        self.assertIn("overwrite=true", calls[0].request.url)
+
+    @responses.activate
+    def test_classification_defaults_to_overwrite(self):
+        self.project.type = "classification"
+        calls, _ = self._upload({"id": "img-id", "duplicate": True})
+        self.assertEqual(len(calls), 1)
+        self.assertIn("overwrite=true", calls[0].request.url)