Merge pull request #830 from superannotateai/FRIDAY-4754

add delete_exports

Author: VaghinakDev

docs/source/api_reference/api_export.rst

@@ -7,3 +7,4 @@ Exports
 .. _ref_download_export:
 .. automethod:: superannotate.SAClient.download_export
 .. automethod:: superannotate.SAClient.get_exports
+.. automethod:: superannotate.SAClient.delete_exports

docs/source/api_reference/helpers.rst

@@ -12,7 +12,6 @@ _________________________________________________________________
 .. _ref_import_annotation_format:
 .. autofunction:: superannotate.import_annotation
 .. autofunction:: superannotate.export_annotation
-.. autofunction:: superannotate.convert_project_type
 
 ----------
 

docs/source/userguide/SDK_Functions_sheet.csv

@@ -81,7 +81,6 @@ Team,get_team_metadata(),Not Relevant,Not Relevant,Not Relevant,Not Relevant,Not
 "Converting
 Annotations",import_annotation(),Not Relevant,Not Relevant,Not Relevant,Not Relevant,Not Relevant
 ,export_annotation(),Not Relevant,Not Relevant,Not Relevant,Not Relevant,Not Relevant
-,convert_project_type(),Not Relevant,Not Relevant,Not Relevant,Not Relevant,Not Relevant
 "Working w/
 Annotations",validate_annotations(),Not Relevant,Not Relevant,Not Relevant,Not Relevant,Not Relevant
 ,aggregate_annotations_as_df(),Not Relevant,Not Relevant,Not Relevant,Not Relevant,Not Relevant

pytest.ini

@@ -3,4 +3,4 @@ minversion = 3.7
 log_cli=true
 python_files = test_*.py
 ;pytest_plugins = ['pytest_profiling']
-addopts = -n 6 --dist loadscope
+;addopts = -n 6 --dist loadscope

src/superannotate/lib/app/interface/sdk_interface.py

@@ -2512,6 +2512,48 @@ def prepare_export(
             raise AppException(response.errors)
         return response.data
 
+    def delete_exports(
+        self,
+        project: Union[NotEmptyStr, int],
+        exports: Union[List[int], List[str], Literal["*"]],
+    ):
+        """Delete one or more exports from the specified project. The exports argument
+        accepts a list of export names or export IDs. The special value “*” means delete all exports.
+
+
+        :param project: The name or ID of the project.
+        :type project: Union[NotEmptyStr, int]
+
+        :param exports: A list of export names or IDs to delete. The special value "*" means delete all exports.
+        :type exports: Union[List[int], List[str], Literal["*"]]
+
+        Request Example:
+        ::
+
+            # To delete a specific export
+            client.delete_exports(
+                project="my_project",
+                exports=["TestProject_Jan_30_2026_12_09"]
+            )
+
+            # To delete all exports in the project
+            client.delete_exports(
+                project="my_project",
+                exports="*"
+            )
+        """
+        project_entity = (
+            self.controller.get_project_by_id(project).data
+            if isinstance(project, int)
+            else self.controller.get_project(project)
+        )
+        response = self.controller.delete_exports(
+            project=project_entity, exports=exports
+        )
+        if response.errors:
+            raise AppException(response.errors)
+        logger.info(f"Successfully removed {response.data} export(s).")
+
     def upload_videos_from_folder_to_project(
         self,
         project: Union[NotEmptyStr, dict],

src/superannotate/lib/core/serviceproviders.py

@@ -872,6 +872,14 @@ def get_export(
     ) -> ServiceResponse:
         raise NotImplementedError
 
+    @abstractmethod
+    def delete_export(
+        self,
+        project: entities.ProjectEntity,
+        export_id: int,
+    ) -> ServiceResponse:
+        raise NotImplementedError
+
     @abstractmethod
     def get_project_images_count(
         self, project: entities.ProjectEntity

src/superannotate/lib/core/usecases/models.py

@@ -7,6 +7,8 @@
 from pathlib import Path
 from tempfile import TemporaryDirectory
 from typing import List
+from typing import Literal
+from typing import Union
 
 import boto3
 import lib.core as constances
@@ -263,6 +265,61 @@ def execute(self):
         return self._response
 
 
+class DeleteExportsUseCase(BaseUseCase):
+    def __init__(
+        self,
+        service_provider: BaseServiceProvider,
+        project: ProjectEntity,
+        exports: Union[List[int], List[str], Literal["*"]],
+    ):
+        super().__init__()
+        self._service_provider = service_provider
+        self._project = project
+        self._exports = exports
+
+    def execute(self):
+        if self.is_valid():
+            deleted_count = 0
+            if self._exports:
+                existing_exports = self._service_provider.get_exports(
+                    self._project
+                ).data
+                export_ids_to_delete = []
+                if existing_exports:
+                    if self._exports == "*":
+                        export_ids_to_delete = [exp["id"] for exp in existing_exports]
+                    else:
+                        # drop duplicates
+                        self._exports = list(set(self._exports))  # noqa
+
+                        if isinstance(self._exports[0], int):
+                            existing_exports_ids = [
+                                int(exp["id"]) for exp in existing_exports
+                            ]
+                            export_ids_to_delete = list(
+                                set(self._exports).intersection(
+                                    set(existing_exports_ids)
+                                )
+                            )
+
+                        elif isinstance(self._exports[0], str):
+                            export_ids_to_delete = [
+                                exp["id"]
+                                for exp in existing_exports
+                                if exp["name"] in self._exports
+                            ]
+
+                    for export_id in export_ids_to_delete:
+                        response = self._service_provider.delete_export(
+                            project=self._project, export_id=export_id
+                        )
+                        if response.ok and response.data.get("success"):
+                            deleted_count += 1
+            self._response.data = deleted_count
+
+        return self._response
+
+
 class ConsensusUseCase(BaseUseCase):
     def __init__(
         self,

src/superannotate/lib/infrastructure/controller.py

@@ -1810,6 +1810,16 @@ def prepare_export(
         )
         return use_case.execute()
 
+    def delete_exports(
+        self, project: ProjectEntity, exports: Union[List[int], List[str], Literal["*"]]
+    ):
+        use_case = usecases.DeleteExportsUseCase(
+            service_provider=self.service_provider,
+            project=project,
+            exports=exports,
+        )
+        return use_case.execute()
+
     def search_team_contributors(self, **kwargs):
         condition = build_condition(**kwargs)
         use_case = usecases.SearchContributorsUseCase(

src/superannotate/lib/infrastructure/serviceprovider.py

@@ -308,6 +308,17 @@ def get_export(self, project: entities.ProjectEntity, export_id: int):
             params={"project_id": project.id},
         )
 
+    def delete_export(
+        self,
+        project: entities.ProjectEntity,
+        export_id: int,
+    ):
+        return self.client.request(
+            f"{self.URL_PREPARE_EXPORT}/{export_id}",
+            "delete",
+            params={"project_id": project.id},
+        )
+
     def get_project_images_count(self, project: entities.ProjectEntity):
         return self.client.request(
             self.URL_FOLDERS_IMAGES,

tests/integration/export/test_export.py

@@ -3,12 +3,14 @@
 import logging
 import os
 import tempfile
+import time
 from unittest import TestCase
 
 import boto3
 from src.superannotate import AppException
 from src.superannotate import SAClient
 from tests import compare_result
+from tests.integration.base import BaseTestCase
 from tests.integration.export import DATA_SET_PATH
 
 sa = SAClient()
@@ -122,3 +124,156 @@ def test_export_with_statuses(self):
             sa.download_export(self.PROJECT_NAME, export, tmpdir_name)
             assert not filecmp.dircmp(tmpdir_name, self.TEST_FOLDER_PATH).left_only
             assert not filecmp.dircmp(tmpdir_name, self.TEST_FOLDER_PATH).right_only
+
+
+class TestDeleteExports(BaseTestCase):
+    PROJECT_NAME = "TestDeleteExports"
+    PROJECT_DESCRIPTION = "Desc"
+    PROJECT_TYPE = "Vector"
+
+    def _check_all_exports_prepared(self, timeout=60):
+        """
+        Wait for all exports to be prepared and return them.
+
+        :param timeout: Maximum time to wait in seconds
+        :return: List of all exports when all are prepared
+        :raises TimeoutError: If exports are not ready within timeout
+        """
+        start_time = time.time()
+        while time.time() - start_time < timeout:
+            exports = sa.get_exports(self.PROJECT_NAME, return_metadata=True)
+
+            # Check if all exports are in a final state (not InProgress)
+            all_ready = all(exp.get("status") == 2 for exp in exports)
+
+            if all_ready:
+                return exports
+
+            time.sleep(2)
+
+        raise TimeoutError("Exports did not complete within the timeout period")
+
+    def test_delete_export_by_name(self):
+        """Test deleting a single export by name"""
+        export1 = sa.prepare_export(self.PROJECT_NAME)
+        exports = self._check_all_exports_prepared()
+        assert len(exports) == 1
+
+        with self.assertLogs("sa", level="INFO") as cm:
+            sa.delete_exports(self.PROJECT_NAME, exports=[export1["name"]])
+            assert "INFO:sa:Successfully removed 1 export(s)." in cm.output[0]
+        exports = sa.get_exports(self.PROJECT_NAME, return_metadata=True)
+        export_names = [exp["name"] for exp in exports]
+
+        assert export1["name"] not in export_names
+
+    def test_delete_export_by_id(self):
+        """Test deleting a single export by ID"""
+        export1 = sa.prepare_export(self.PROJECT_NAME)
+        exports = self._check_all_exports_prepared()
+        assert len(exports) == 1
+
+        export2 = sa.prepare_export(self.PROJECT_NAME)
+        exports = self._check_all_exports_prepared()
+        assert len(exports) == 2
+
+        sa.delete_exports(self.PROJECT_NAME, exports=[export1["id"]])
+        exports = sa.get_exports(self.PROJECT_NAME, return_metadata=True)
+        export_ids = [exp["id"] for exp in exports]
+
+        assert export1["id"] not in export_ids
+        assert export2["id"] in export_ids
+
+    def test_delete_multiple_exports(self):
+        """Test deleting multiple exports by name and ID"""
+        export1 = sa.prepare_export(self.PROJECT_NAME)
+        exports = self._check_all_exports_prepared()
+        assert len(exports) == 1
+
+        export2 = sa.prepare_export(self.PROJECT_NAME)
+        exports = self._check_all_exports_prepared()
+        assert len(exports) == 2
+
+        export3 = sa.prepare_export(self.PROJECT_NAME)
+        exports = self._check_all_exports_prepared()
+        assert len(exports) == 3
+
+        sa.delete_exports(self.PROJECT_NAME, exports=[export1["id"], export2["id"]])
+
+        exports = sa.get_exports(self.PROJECT_NAME, return_metadata=True)
+        export_ids = [exp["id"] for exp in exports]
+
+        assert export1["id"] not in export_ids
+        assert export2["id"] not in export_ids
+        assert export3["id"] in export_ids
+
+    def test_delete_all_exports(self):
+        """Test deleting all exports using '*'"""
+        export1 = sa.prepare_export(self.PROJECT_NAME)
+        exports = self._check_all_exports_prepared()
+        assert len(exports) == 1
+
+        export2 = sa.prepare_export(self.PROJECT_NAME)
+        exports = self._check_all_exports_prepared()
+        assert len(exports) == 2
+
+        sa.delete_exports(self.PROJECT_NAME, exports="*")
+
+        exports = sa.get_exports(self.PROJECT_NAME, return_metadata=True)
+        assert len(exports) == 0
+
+    def test_delete_nonexistent_export(self):
+        """Test deleting a non-existent export (should not raise error)"""
+        export1 = sa.prepare_export(self.PROJECT_NAME)
+        exports = self._check_all_exports_prepared()
+        assert len(exports) == 1
+
+        # Should not raise error for non-existent export
+        with self.assertLogs("sa", level="INFO") as cm:
+            sa.delete_exports(
+                self.PROJECT_NAME, exports=["nonexistent_export", export1["name"]]
+            )
+            assert "Successfully removed 1 export(s)." in cm.output[0]
+
+        exports = sa.get_exports(self.PROJECT_NAME, return_metadata=True)
+        export_names = [exp["name"] for exp in exports]
+
+        assert export1["name"] not in export_names
+
+    def test_delete_exports_empty_list(self):
+        """Test deleting with empty list"""
+        export1 = sa.prepare_export(self.PROJECT_NAME)
+        exports = self._check_all_exports_prepared()
+        assert len(exports) == 1
+
+        with self.assertLogs("sa", level="INFO") as cm:
+            sa.delete_exports(self.PROJECT_NAME, exports=[])
+            assert "Successfully removed 0 export(s)." in cm.output[0]
+
+        exports = sa.get_exports(self.PROJECT_NAME, return_metadata=True)
+        assert len(exports) == 1
+
+    def test_delete_exports_project_not_found(self):
+        """Test deleting exports from non-existent project"""
+        with self.assertRaisesRegexp(AppException, "Project not found"):
+            sa.delete_exports("NonExistentProject123456", exports=["*"])
+
+    def test_delete_mixed_valid_invalid_exports(self):
+        """Test deleting mix of valid and invalid export identifiers"""
+        export1 = sa.prepare_export(self.PROJECT_NAME)
+        exports = self._check_all_exports_prepared()
+        assert len(exports) == 1
+
+        export2 = sa.prepare_export(self.PROJECT_NAME)
+        exports = self._check_all_exports_prepared()
+        assert len(exports) == 2
+
+        with self.assertLogs("sa", level="INFO") as cm:
+            sa.delete_exports(
+                self.PROJECT_NAME,
+                exports=[export1["name"], "invalid_name", 99999, export2["id"]],
+            )
+            assert "Successfully removed 2 export(s)." in cm.output[0]
+
+        exports = sa.get_exports(self.PROJECT_NAME, return_metadata=True)
+        assert len(exports) == 0