feat: annotation-overwrite parameter on upload_dataset and the CLI

roboflow/adapters/rfapi.py

@@ -897,10 +897,12 @@ def _save_annotation_error(response):
 # ---------------------------------------------------------------------------
 
 
-def init_zip_upload(api_key, workspace_url, project_url, split=None, tags=None, batch_name=None) -> dict:
+def init_zip_upload(
+    api_key, workspace_url, project_url, split=None, tags=None, batch_name=None, annotation_overwrite=False
+) -> dict:
     """POST /{ws}/{proj}/upload/zip — initialize a zip upload and get a signed URL."""
     url = f"{API_URL}/{workspace_url}/{project_url}/upload/zip"
-    body: Dict[str, Union[str, List[str]]] = {}
+    body: Dict[str, Union[str, List[str], bool]] = {"annotationOverwrite": annotation_overwrite}
     if split is not None:
         body["split"] = split
     if tags is not None:

roboflow/cli/handlers/image.py

@@ -40,6 +40,14 @@ def upload_image(
         bool,
         typer.Option("--zip-upload", help="Zip the directory client-side and use the async zip upload flow"),
     ] = False,
+    annotation_overwrite: Annotated[
+        Optional[bool],
+        typer.Option(
+            "--annotation-overwrite/--no-annotation-overwrite",
+            help="Zip flow: overwrite existing annotations on duplicate images "
+            "(default: off, except classification projects where the API requires it on)",
+        ),
+    ] = None,
     no_wait: Annotated[
         bool,
         typer.Option("--no-wait", help="Zip flow: return immediately with task_id instead of polling"),
@@ -60,6 +68,7 @@ def upload_image(
         labelmap=labelmap,
         is_prediction=is_prediction,
         zip_upload=zip_upload,
+        annotation_overwrite=annotation_overwrite,
         no_wait=no_wait,
     )
     _handle_upload(args)
@@ -348,6 +357,7 @@ def _handle_upload_directory(args, api_key: str, path: str) -> None:  # noqa: AN
             num_retries=retries,
             is_prediction=getattr(args, "is_prediction", False),
             use_zip_upload=getattr(args, "zip_upload", False),
+            annotation_overwrite=getattr(args, "annotation_overwrite", None),
             split=getattr(args, "split", None),
             tags=tags,
             wait=wait,

roboflow/core/workspace.py

@@ -514,6 +514,7 @@ def upload_dataset(
         is_prediction=False,
         *,
         use_zip_upload: bool = False,
+        annotation_overwrite: Optional[bool] = None,
         tags: Optional[List[str]] = None,
         split: Optional[str] = None,
         wait: bool = True,
@@ -538,6 +539,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).
             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.
@@ -576,13 +578,17 @@ 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,
                     project_slug,
                     split=split,
                     tags=tags,
                     batch_name=batch_name,
+                    annotation_overwrite=annotation_overwrite,
                 )
                 print(f"Uploading zip to Roboflow (task_id={init['taskId']})...")
                 rfapi.upload_zip_to_signed_url(init["signedUrl"], zip_path)

tests/cli/test_image_handler.py

@@ -345,6 +345,61 @@ def test_zip_upload_flag_defaults_false(self, mock_rf_cls):
             _, kwargs = mock_ws.upload_dataset.call_args
             self.assertEqual(kwargs.get("use_zip_upload"), False)
 
+    @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)
+
+    @patch("roboflow.Roboflow")
+    def test_annotation_overwrite_forwarded_to_upload_dataset(self, mock_rf_cls):
+        from roboflow.cli.handlers.image import _handle_upload
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            mock_ws = MagicMock()
+            mock_ws.upload_dataset.return_value = {"status": "completed", "task_id": "t1"}
+            mock_rf_cls.return_value.workspace.return_value = mock_ws
+
+            args = _make_args(
+                json=True,
+                path=tmpdir,
+                project="proj",
+                annotation=None,
+                split=None,
+                batch=None,
+                tag=None,
+                metadata=None,
+                concurrency=10,
+                retries=0,
+                labelmap=None,
+                is_prediction=False,
+                zip_upload=True,
+                annotation_overwrite=True,
+                no_wait=False,
+            )
+
+            buf = io.StringIO()
+            old = sys.stdout
+            sys.stdout = buf
+            try:
+                _handle_upload(args)
+            finally:
+                sys.stdout = old
+
+            _, kwargs = mock_ws.upload_dataset.call_args
+            self.assertEqual(kwargs.get("annotation_overwrite"), True)
+
     @patch("roboflow.Roboflow")
     def test_upload_directory_omits_default_split_when_not_explicit(self, mock_rf_cls):
         from roboflow.cli.handlers.image import _handle_upload

tests/test_project.py

@@ -1021,6 +1021,74 @@ def test_directory_with_use_zip_upload_zips_and_cleans_up(self):
             if _os.path.isdir(src_dir):
                 _os.rmdir(src_dir)
 
+    def test_annotation_overwrite_defaults_false(self):
+        import tempfile
+
+        with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as fh:
+            fh.write(b"fake zip")
+            zip_path = fh.name
+
+        mocks = self._rfapi_mocks()
+        started = {name: m.start() for name, m in mocks.items()}
+        try:
+            self.workspace.upload_dataset(dataset_path=zip_path, project_name=PROJECT_NAME)
+            _, kwargs = started["init"].call_args
+            self.assertEqual(kwargs.get("annotation_overwrite"), False)
+        finally:
+            for m in mocks.values():
+                m.stop()
+            import os as _os
+
+            if _os.path.exists(zip_path):
+                _os.unlink(zip_path)
+
+    def test_annotation_overwrite_defaults_true_for_classification(self):
+        import tempfile
+        from types import SimpleNamespace
+
+        with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as fh:
+            fh.write(b"fake zip")
+            zip_path = fh.name
+
+        mocks = self._rfapi_mocks()
+        mocks["project"] = patch(
+            "roboflow.core.workspace.Workspace._get_or_create_project",
+            return_value=(SimpleNamespace(id=f"{WORKSPACE_NAME}/{PROJECT_NAME}", type="classification"), False),
+        )
+        started = {name: m.start() for name, m in mocks.items()}
+        try:
+            self.workspace.upload_dataset(dataset_path=zip_path, project_name=PROJECT_NAME)
+            _, kwargs = started["init"].call_args
+            self.assertEqual(kwargs.get("annotation_overwrite"), True)
+        finally:
+            for m in mocks.values():
+                m.stop()
+            import os as _os
+
+            if _os.path.exists(zip_path):
+                _os.unlink(zip_path)
+
+    def test_annotation_overwrite_explicit_passthrough(self):
+        import tempfile
+
+        with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as fh:
+            fh.write(b"fake zip")
+            zip_path = fh.name
+
+        mocks = self._rfapi_mocks()
+        started = {name: m.start() for name, m in mocks.items()}
+        try:
+            self.workspace.upload_dataset(dataset_path=zip_path, project_name=PROJECT_NAME, annotation_overwrite=True)
+            _, kwargs = started["init"].call_args
+            self.assertEqual(kwargs.get("annotation_overwrite"), True)
+        finally:
+            for m in mocks.values():
+                m.stop()
+            import os as _os
+
+            if _os.path.exists(zip_path):
+                _os.unlink(zip_path)
+
     def test_directory_default_stays_on_per_image(self):
         import tempfile