[PT-5400] add bulk deletion (#762)

---------

Co-authored-by: Matheus Pinheiro <[email protected]>

Author: jmigueldelgado

CHANGELOG.md

@@ -30,6 +30,9 @@ You can check your current version with the following command:
 
 For more information, see [UP42 Python package description](https://pypi.org/project/up42-py/).
 
+### 2.5.0a4
+**August 6, 2025**
+- Added bulk deletion of items: `BulkDeletion`.
 
 ### 2.5.0a3
 **Aug 4, 2025**

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "up42-py"
-version = "2.5.0a3"
+version = "2.5.0a4"
 description = "Python SDK for UP42, the geospatial marketplace and developer platform."
 authors = ["UP42 GmbH <[email protected]>"]
 license = "https://github.com/up42/up42-py/blob/master/LICENSE"

tests/test_stac.py

@@ -1,5 +1,6 @@
 import datetime as dt
 import uuid
+from unittest import mock
 
 import pystac
 import pytest
@@ -143,3 +144,49 @@ def test_should_set_up42_extension(self, entity, entity_dict, attribute, key):
         new_value = "new-value"
         setattr(entity.up42, attribute, new_value)  # type: ignore
         assert entity_dict[key] == new_value
+
+
+class TestBulkDeletion:
+    items = [
+        pystac.Item(
+            id=f"item-id-{i}",
+            collection="collection-id",
+            geometry=None,
+            bbox=None,
+            datetime=dt.datetime.now(),
+            properties={},
+        )
+        for i in range(2)
+    ]
+    collection_with_all_items = pystac.Collection(
+        id="collection-id",
+        description="",
+        extent=pystac.Extent(
+            spatial=pystac.SpatialExtent(bboxes=[[1.0, 2.0, 3.0, 4.0]]),
+            temporal=pystac.TemporalExtent(intervals=[[dt.datetime.now(), None]]),
+        ),
+        extra_fields={},
+    )
+    collection_with_all_items.add_items(items)
+
+    def test_should_raise_and_not_submit_when_missing_items(self):
+        mock_stac_client = mock.Mock()
+        mock_stac_client.get_items.return_value = iter([self.items[0]])
+        bulk_deletion = stac.BulkDeletion(self.items[0].id)
+        bulk_deletion.stac_client = mock_stac_client
+        with pytest.raises(stac.IncompleteCollectionDeletionError):
+            bulk_deletion.delete()
+
+    def test_should_delete_staged_items(self, requests_mock: req_mock.Mocker):
+        requests_mock.delete(
+            url=f"/v2/assets/stac/collections/{self.collection_with_all_items.id}",
+            status_code=204,
+        )
+        mock_stac_client = mock.Mock()
+        mock_stac_client.get_items.return_value = iter(self.items)
+
+        bulk_deletion = stac.BulkDeletion(self.items[0].id, self.items[1].id)
+        bulk_deletion.stac_client = mock_stac_client
+
+        bulk_deletion.delete()
+        assert requests_mock.request_history[0].method == "DELETE"

up42/__init__.py

@@ -34,6 +34,7 @@
 from up42.order import Order, OrderSorting
 from up42.order_template import BatchOrderTemplate
 from up42.processing import Job, JobSorting, JobStatus
+from up42.stac import BulkDeletion
 from up42.stac import extend as stac_extend
 from up42.storage import Storage
 from up42.tasking import FeasibilityStudy, FeasibilityStudySorting, Quotation, QuotationSorting, Tasking
@@ -83,5 +84,6 @@
         QuotationSorting,
         FeasibilityStudy,
         FeasibilityStudySorting,
+        BulkDeletion,
     ]
 ]

up42/stac.py

@@ -9,6 +9,12 @@ class InvalidUp42Asset(ValueError):
     pass
 
 
+class IncompleteCollectionDeletionError(ValueError):
+    """Raised when attempting to delete a collection but not all items in the collection are included."""
+
+    pass
+
+
 class FileProvider:
     session = base.Session()
 
@@ -96,3 +102,34 @@ def extend():
 
     pystac.Item.up42 = Up42ExtensionProvider()  # type: ignore
     pystac.Collection.up42 = Up42ExtensionProvider()  # type: ignore
+
+
+class BulkDeletion:
+    session = base.Session()
+    stac_client = base.StacClient()
+
+    def __init__(self, *item_ids: str):
+        self._item_ids = set(item_ids)
+
+    def delete(self):
+        items = self.stac_client.get_items(*self._item_ids)
+        collections: set[pystac.Collection] = set()
+        for item in items:
+            collection = item.get_parent()
+            collections.add(collection)  # type: ignore
+
+        for collection in collections:
+            collection_item_ids = {item_in_collection.id for item_in_collection in collection.get_items()}
+            missing_items = collection_item_ids - self._item_ids
+            if missing_items:
+                error_msg = (
+                    f"The deletion request failed because the submitted items are part of a group that must be deleted "
+                    f"together. The submitted items belong to the following collection: '{collection.id}'. All items "
+                    f"from this collection must be deleted at once. To proceed, please add these missing item IDs "
+                    f"to your request: {list(missing_items)}."
+                )
+                raise IncompleteCollectionDeletionError(error_msg)
+
+        for collection in collections:
+            url = host.endpoint(f"/v2/assets/stac/collections/{collection.id}")
+            self.session.delete(url=url)