Merge pull request #22180 from mvdbeek/batch-dataset-purge-task

[26.0] Add batch celery task for history dataset purging

client/src/api/schema/schema.ts

@@ -7463,6 +7463,11 @@ export interface components {
              * @description Whether this resource is currently publicly available to all users.
              */
             published: boolean;
+            /**
+             * Purge Task
+             * @description Summary of the async task purging datasets in this history. Only present when purge is performed via a background task.
+             */
+            purge_task?: components["schemas"]["AsyncTaskResultSummary"] | null;
             /**
              * Purged
              * @description Whether this item has been permanently removed.
@@ -7586,6 +7591,11 @@ export interface components {
              * @description Whether this resource is currently publicly available to all users.
              */
             published: boolean;
+            /**
+             * Purge Task
+             * @description Summary of the async task purging datasets in this history. Only present when purge is performed via a background task.
+             */
+            purge_task?: components["schemas"]["AsyncTaskResultSummary"] | null;
             /**
              * Purged
              * @description Whether this item has been permanently removed.
@@ -9974,6 +9984,11 @@ export interface components {
              * @description Whether this resource is currently publicly available to all users.
              */
             published?: boolean | null;
+            /**
+             * Purge Task
+             * @description Summary of the async task purging datasets in this history. Only present when purge is performed via a background task.
+             */
+            purge_task?: components["schemas"]["AsyncTaskResultSummary"] | null;
             /**
              * Purged
              * @description Whether this item has been permanently removed.
@@ -10213,6 +10228,11 @@ export interface components {
              * @description Whether this resource is currently publicly available to all users.
              */
             published?: boolean | null;
+            /**
+             * Purge Task
+             * @description Summary of the async task purging datasets in this history. Only present when purge is performed via a background task.
+             */
+            purge_task?: components["schemas"]["AsyncTaskResultSummary"] | null;
             /**
              * Purged
              * @description Whether this item has been permanently removed.
@@ -15492,6 +15512,11 @@ export interface components {
              * @description Whether this resource is currently publicly available to all users.
              */
             published: boolean;
+            /**
+             * Purge Task
+             * @description Summary of the async task purging datasets in this history. Only present when purge is performed via a background task.
+             */
+            purge_task?: components["schemas"]["AsyncTaskResultSummary"] | null;
             /**
              * Purged
              * @description Whether this item has been permanently removed.
@@ -15613,6 +15638,11 @@ export interface components {
              * @description Whether this resource is currently publicly available to all users.
              */
             published: boolean;
+            /**
+             * Purge Task
+             * @description Summary of the async task purging datasets in this history. Only present when purge is performed via a background task.
+             */
+            purge_task?: components["schemas"]["AsyncTaskResultSummary"] | null;
             /**
              * Purged
              * @description Whether this item has been permanently removed.

lib/galaxy/celery/tasks.py

@@ -11,8 +11,11 @@ from typing import (
 )
 
 from sqlalchemy import (
+    and_,
     exists,
+    false,
     select,
+    update,
 )
 
 from galaxy import model
@@ -58,6 +61,7 @@ from galaxy.schema.tasks import (
     MaterializeDatasetInstanceTaskRequest,
     PrepareDatasetCollectionDownload,
     PurgeDatasetsTaskRequest,
+    PurgeHistoryDatasetsTaskRequest,
     QueueJobs,
     SetupHistoryExportJob,
     TOOL_SOURCE_CLASS,
@@ -117,6 +121,57 @@ def purge_datasets(
     dataset_manager.purge_datasets(request)
 
 
+@galaxy_task(action="purge all datasets in a history")
+def purge_history_datasets(
+    sa_session: galaxy_scoped_session,
+    dataset_manager: DatasetManager,
+    object_store: BaseObjectStore,
+    request: PurgeHistoryDatasetsTaskRequest,
+    task_user_id: Optional[int] = None,
+):
+    """Batch purge all HDAs in a history in a single task.
+
+    Bulk-marks all unpurged HDAs as deleted/purged, recalculates user quota,
+    and removes underlying dataset files from the object store.
+    """
+    history = sa_session.get(model.History, request.history_id)
+    if not history:
+        log.error(f"Purge history datasets task failed, history {request.history_id} not found")
+        return
+    # Collect dataset IDs before the bulk update
+    dataset_id_stmt = (
+        select(model.HistoryDatasetAssociation.dataset_id)
+        .where(
+            and_(
+                model.HistoryDatasetAssociation.history_id == request.history_id,
+                model.HistoryDatasetAssociation.purged == false(),
+            )
+        )
+        .distinct()
+    )
+    dataset_ids = list(sa_session.scalars(dataset_id_stmt))
+    if not dataset_ids:
+        return
+    # Bulk mark all unpurged HDAs as deleted and purged
+    sa_session.execute(
+        update(model.HistoryDatasetAssociation)
+        .where(
+            and_(
+                model.HistoryDatasetAssociation.history_id == request.history_id,
+                model.HistoryDatasetAssociation.purged == false(),
+            )
+        )
+        .values(deleted=True, purged=True)
+    )
+    sa_session.commit()
+    # Recalculate user disk usage from scratch
+    user = history.user
+    if user:
+        user.calculate_and_set_disk_usage(object_store)
+    # Remove underlying dataset files from object store
+    dataset_manager.purge_datasets(PurgeDatasetsTaskRequest(dataset_ids=dataset_ids))
+
+
 @galaxy_task(ignore_result=True, action="materializing dataset instance")
 def materialize(
     hda_manager: HDAManager,

lib/galaxy/managers/histories.py

@@ -79,6 +79,7 @@ from galaxy.schema.storage_cleaner import (
     StoredItem,
     StoredItemOrderBy,
 )
+from galaxy.schema.tasks import PurgeHistoryDatasetsTaskRequest
 from galaxy.security.validate_user_input import validate_preferred_object_store_id
 from galaxy.structured_app import MinimalManagerApp
 from galaxy.util.search import (
@@ -292,12 +293,21 @@ class HistoryManager(sharable.SharableModelManager[model.History], deletable.Pur
         self.error_unless_mutable(item)
         self.hda_manager.dataset_manager.error_unless_dataset_purge_allowed()
         # First purge all the datasets
-        for hda in item.datasets:
-            if not hda.purged:
-                self.hda_manager.purge(hda, flush=True, **kwargs)
+        if self.app.config.enable_celery_tasks:
+            from galaxy.celery.tasks import purge_history_datasets
+
+            request = PurgeHistoryDatasetsTaskRequest(history_id=item.id)
+            user = item.user
+            result = purge_history_datasets.delay(request=request, task_user_id=user.id if user else None)
+        else:
+            result = None
+            for hda in item.datasets:
+                if not hda.purged:
+                    self.hda_manager.purge(hda, flush=True, **kwargs)
 
         # Now mark the history as purged
         super().purge(item, flush=flush, **kwargs)
+        return result
 
     # .... current
     # TODO: make something to bypass the anon user + current history permissions issue

lib/galaxy/schema/schema.py

@@ -1435,6 +1435,11 @@ class HistorySummary(Model, WithModelClass):
     tags: TagCollection
     update_time: datetime = UpdateTimeField
     preferred_object_store_id: Optional[str] = PreferredObjectStoreIdField
+    purge_task: Optional["AsyncTaskResultSummary"] = Field(
+        None,
+        title="Purge Task",
+        description="Summary of the async task purging datasets in this history. Only present when purge is performed via a background task.",
+    )
 
 
 class HistoryActiveContentCounts(Model):
@@ -3997,6 +4002,13 @@ class AsyncTaskResultSummary(Model):
     )
 
 
+HistorySummary.model_rebuild()
+HistoryDetailed.model_rebuild()
+CustomHistoryView.model_rebuild()
+ArchivedHistorySummary.model_rebuild()
+ArchivedHistoryDetailed.model_rebuild()
+CustomArchivedHistoryView.model_rebuild()
+
 ToolRequestIdField = Field(title="ID", description="Encoded ID of the role")
 
 

lib/galaxy/schema/tasks.py

@@ -139,6 +139,10 @@ class PurgeDatasetsTaskRequest(Model):
     dataset_ids: list[int]
 
 
+class PurgeHistoryDatasetsTaskRequest(Model):
+    history_id: int
+
+
 class TaskState(str, Enum):
     """Enum representing the possible states of a task."""
 

lib/galaxy/webapps/galaxy/services/histories.py

@@ -454,10 +454,14 @@ class HistoriesService(ServiceBase, ConsumesModelStores, ServesExportStores):
         """
         history = self.manager.get_mutable(history_id, trans.user, current_history=trans.history)
         if purge:
-            self.manager.purge(history)
+            result = self.manager.purge(history)
         else:
+            result = None
             self.manager.delete(history)
-        return self._serialize_history(trans, history, serialization_params)
+        rval = self._serialize_history(trans, history, serialization_params)
+        if result is not None:
+            rval["purge_task"] = async_task_summary(result)
+        return rval
 
     def undelete(
         self,

test/integration/test_purge_datasets.py

@@ -76,6 +76,43 @@ class TestPurgeDatasetsIntegration(integration_util.IntegrationTestCase):
         purge_result = purge_response.json()
         assert purge_result["success_count"] == 1
 
+    def test_purge_history_removes_underlying_datasets_from_disk(self):
+        """Test that purging a history purges all its datasets and removes files from disk."""
+        hda1 = self.dataset_populator.new_dataset(self.test_history_id, wait=True)
+        hda2 = self.dataset_populator.new_dataset(self.test_history_id, wait=True)
+        hda1_id = hda1["id"]
+        hda2_id = hda2["id"]
+
+        # Ensure dataset files exist on disk
+        dataset_file1 = self._get_underlying_dataset_on_disk(hda1_id)
+        dataset_file2 = self._get_underlying_dataset_on_disk(hda2_id)
+        assert self._file_exists_on_disk(dataset_file1)
+        assert self._file_exists_on_disk(dataset_file2)
+
+        # Purge the entire history
+        purge_response = self._delete(f"histories/{self.test_history_id}", data={"purge": True}, json=True)
+        self._assert_status_code_is_ok(purge_response)
+        purge_result = purge_response.json()
+
+        # Verify history is purged
+        assert purge_result["purged"]
+        assert purge_result["deleted"]
+
+        # Verify the response contains a purge_task with an id
+        assert "purge_task" in purge_result
+        purge_task = purge_result["purge_task"]
+        assert "id" in purge_task
+        purge_task_id = purge_task["id"]
+
+        # Wait for the celery task to complete via the tasks API
+        self.dataset_populator.wait_on_task_id(purge_task_id)
+
+        # After task completion, HDAs should be purged and files deleted
+        self.dataset_populator.wait_for_purge(self.test_history_id, hda1_id)
+        self.dataset_populator.wait_for_purge(self.test_history_id, hda2_id)
+        assert not self._file_exists_on_disk(dataset_file1)
+        assert not self._file_exists_on_disk(dataset_file2)
+
     def _get_underlying_dataset_on_disk(self, hda_id: str) -> Optional[str]:
         detailed_response = self._get(f"datasets/{hda_id}", admin=True).json()
         return detailed_response.get("file_name")