opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-21 13:50:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

Rename API and service methods to clarify bulk storage operations

Browse Source
This commit is contained in:
davelopez
2026-05-20 18:52:33 +02:00
parent 9a90ebe62b
commit ebd54d8c1b
4 changed files with 58 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/webapps/galaxy/api/history_contents.py
+8 -8
View File
@@ -842,44 +842,44 @@ class FastAPIHistoryContents:
"/api/histories/{history_id}/contents/bulk/storage/preview",
summary="Previews a storage bulk operation for selected history contents.",
)
def storage_operation_preview(
def bulk_storage_operation_preview(
self,
history_id: HistoryIDPathParam,
trans: ProvidesHistoryContext = DependsOnTrans,
filter_query_params: ValueFilterQueryParams = Depends(get_value_filter_query_params),
payload: StorageOperationPreviewRequest = Body(...),
) -> StorageOperationPreviewResponse:
return self.service.storage_operation_preview(trans, history_id, filter_query_params, payload)
return self.service.bulk_storage_operation_preview(trans, history_id, filter_query_params, payload)
@router.post(
"/api/histories/{history_id}/contents/bulk/storage/execute",
summary="Executes a previously previewed storage bulk operation snapshot.",
)
def storage_operation_execute(
def bulk_storage_operation_execute(
self,
history_id: HistoryIDPathParam,
trans: ProvidesHistoryContext = DependsOnTrans,
payload: StorageOperationExecuteRequest = Body(...),
) -> StorageOperationExecuteResponse:
return self.service.storage_operation_execute(trans, history_id, payload)
return self.service.bulk_storage_operation_execute(trans, history_id, payload)
@router.get(
"/api/histories/{history_id}/contents/bulk/storage/runs/{run_id}",
summary="Returns run status summary for a storage bulk operation.",
)
def storage_operation_run(
def bulk_storage_operation_run(
self,
history_id: HistoryIDPathParam,
run_id: DecodedDatabaseIdField,
trans: ProvidesHistoryContext = DependsOnTrans,
) -> StorageOperationRunResponse:
return self.service.storage_operation_run(trans, history_id, run_id)
return self.service.bulk_storage_operation_run(trans, history_id, run_id)
@router.get(
"/api/histories/{history_id}/contents/bulk/storage/runs/{run_id}/items",
summary="Returns paginated per-item details for a storage bulk operation run.",
)
def storage_operation_run_items(
def bulk_storage_operation_run_items(
self,
history_id: HistoryIDPathParam,
run_id: DecodedDatabaseIdField,
@@ -889,7 +889,7 @@ class FastAPIHistoryContents:
limit: int = StorageRunLimitQueryParam,
search: Optional[str] = StorageRunSearchQueryParam,
) -> list[StorageOperationRunItemStatus]:
run_items, total_matches = self.service.storage_operation_run_items(
run_items, total_matches = self.service.bulk_storage_operation_run_items(
trans,
history_id,
run_id,
lib/galaxy/webapps/galaxy/services/history_contents.py
+3 -3
View File
@@ -750,7 +750,7 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto
success_count = len(contents) - len(errors)
return HistoryContentBulkOperationResult(success_count=success_count, errors=errors)
def storage_operation_preview(
def bulk_storage_operation_preview(
self,
trans: ProvidesHistoryContext,
history_id: DecodedDatabaseIdField,
@@ -809,7 +809,7 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto
trans.sa_session.commit()
return StorageOperationExecuteResponse(run=self.storage_operation_manager.to_run_summary(run))
def storage_operation_run(
def bulk_storage_operation_run(
self,
trans: ProvidesHistoryContext,
history_id: DecodedDatabaseIdField,
@@ -826,7 +826,7 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto
summary = self.storage_operation_manager.to_run_summary(run)
return StorageOperationRunResponse(run=summary)
def storage_operation_run_items(
def bulk_storage_operation_run_items(
self,
trans: ProvidesHistoryContext,
history_id: DecodedDatabaseIdField,
lib/galaxy_test/base/populators.py
+28 -12
View File
@@ -1504,7 +1504,9 @@ class BaseDatasetPopulator(BasePopulator):
get_storage_response = self._get(storage_url)
return get_storage_response
def storage_preview_raw(self, history_id: str, payload: dict[str, Any], expected_status: int = 200) -> Response:
def bulk_storage_operation_preview_raw(
self, history_id: str, payload: dict[str, Any], expected_status: int = 200
) -> Response:
response = self._post(
f"histories/{history_id}/contents/bulk/storage/preview",
data=payload,
@@ -1513,10 +1515,14 @@ class BaseDatasetPopulator(BasePopulator):
api_asserts.assert_status_code_is(response, expected_status)
return response
def storage_preview(self, history_id: str, payload: dict[str, Any], expected_status: int = 200) -> dict[str, Any]:
return self.storage_preview_raw(history_id, payload, expected_status=expected_status).json()
def bulk_storage_operation_preview(
self, history_id: str, payload: dict[str, Any], expected_status: int = 200
) -> dict[str, Any]:
return self.bulk_storage_operation_preview_raw(history_id, payload, expected_status=expected_status).json()
def storage_execute_raw(self, history_id: str, payload: dict[str, Any], expected_status: int = 200) -> Response:
def bulk_storage_operation_execute_raw(
self, history_id: str, payload: dict[str, Any], expected_status: int = 200
) -> Response:
response = self._post(
f"histories/{history_id}/contents/bulk/storage/execute",
data=payload,
@@ -1525,15 +1531,25 @@ class BaseDatasetPopulator(BasePopulator):
api_asserts.assert_status_code_is(response, expected_status)
return response
def storage_execute(self, history_id: str, payload: dict[str, Any], expected_status: int = 200) -> dict[str, Any]:
return self.storage_execute_raw(history_id, payload, expected_status=expected_status).json()
def bulk_storage_operation_execute(
self, history_id: str, payload: dict[str, Any], expected_status: int = 200
) -> dict[str, Any]:
return self.bulk_storage_operation_execute_raw(history_id, payload, expected_status=expected_status).json()
def storage_run_status(self, history_id: str, run_id: str, expected_status: int = 200) -> dict[str, Any]:
def bulk_storage_operation_run_status_raw(
self, history_id: str, run_id: str, expected_status: int = 200
) -> Response:
response = self._get(f"histories/{history_id}/contents/bulk/storage/runs/{run_id}")
api_asserts.assert_status_code_is(response, expected_status)
return response
def bulk_storage_operation_run_status(
self, history_id: str, run_id: str, expected_status: int = 200
) -> dict[str, Any]:
response = self.bulk_storage_operation_run_status_raw(history_id, run_id, expected_status=expected_status)
return response.json()
def storage_run_items(
def bulk_storage_operation_run_items(
self,
history_id: str,
run_id: str,
@@ -1552,7 +1568,7 @@ class BaseDatasetPopulator(BasePopulator):
api_asserts.assert_status_code_is(response, expected_status)
return response.json()
def wait_for_storage_run(
def wait_for_bulk_storage_operation_run(
self,
history_id: str,
run_id: str,
@@ -1561,14 +1577,14 @@ class BaseDatasetPopulator(BasePopulator):
timeout: timeout_type = DEFAULT_TIMEOUT,
) -> dict[str, Any]:
def is_terminal():
run_data = self.storage_run_status(history_id, run_id)
run_data = self.bulk_storage_operation_run_status(history_id, run_id)
if run_data["run"]["state"] in ("completed", "failed"):
if include_items_on_terminal:
run_data["items"] = self.storage_run_items(history_id, run_id, search=search)
run_data["items"] = self.bulk_storage_operation_run_items(history_id, run_id, search=search)
return run_data
return None
return wait_on(is_terminal, "storage run completion", timeout=timeout)
return wait_on(is_terminal, "bulk storage operation run completion", timeout=timeout)
def get_roles(self) -> list:
using_requirement("admin")
test/integration/objectstore/test_bulk_storage_operations.py
+19 -14
View File
@@ -36,7 +36,10 @@ from typing import (
from unittest.mock import patch
from uuid import uuid4
from galaxy.celery.tasks import recover_stale_storage_operation_runs
from galaxy.celery.tasks import (
prune_expired_storage_operations,
recover_stale_storage_operation_runs,
)
from galaxy.managers.dataset_storage_operations import (
DatasetStorageOperationManager,
StorageOperationRunExecutor,
@@ -171,7 +174,7 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase):
items: list[dict[str, Any]],
expected_status: int = 200,
) -> dict[str, Any]:
return self.dataset_populator.storage_preview(
return self.dataset_populator.bulk_storage_operation_preview(
history_id,
{
"mode": "move",
@@ -188,7 +191,7 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase):
skip_ineligible: bool = True,
expected_status: int = 200,
) -> dict[str, Any]:
return self.dataset_populator.storage_execute(
return self.dataset_populator.bulk_storage_operation_execute(
history_id,
{
"snapshot_id": snapshot_id,
@@ -213,7 +216,7 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase):
skip_ineligible=skip_ineligible,
)
run = execute_result["run"]
final = self.dataset_populator.wait_for_storage_run(
final = self.dataset_populator.wait_for_bulk_storage_operation_run(
history_id,
run["run_id"],
include_items_on_terminal=include_items_on_terminal,
@@ -222,7 +225,7 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase):
return preview, run, final
def _run_items(self, history_id: str, run_id: str, search: str | None = None) -> list[dict[str, Any]]:
return self.dataset_populator.storage_run_items(history_id, run_id, search=search)
return self.dataset_populator.bulk_storage_operation_run_items(history_id, run_id, search=search)
def _run_items_by_dataset_id(self, items: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
return {item["dataset_id"]: item for item in items}
@@ -269,7 +272,7 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase):
else:
executor.execute_run(snapshot)
return self._app.security.encode_id(run.id)
return self._encode_id(run.id)
def _item(self, hda_id: str) -> dict[str, Any]:
return {"id": hda_id, "history_content_type": "dataset"}
@@ -746,7 +749,9 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase):
def test_get_run_unknown_id_returns_404(self):
"""GET on an unknown run_id returns HTTP 404."""
with self.dataset_populator.test_history() as history_id:
self.dataset_populator.storage_run_status(history_id, self._unknown_encoded_id(), expected_status=404)
self.dataset_populator.bulk_storage_operation_run_status(
history_id, self._unknown_encoded_id(), expected_status=404
)
# ------------------------------------------------------------------ collection (HDCA) tests
@@ -819,7 +824,7 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase):
skip_ineligible=False,
)
run = execute_result["run"]
final = self.dataset_populator.wait_for_storage_run(
final = self.dataset_populator.wait_for_bulk_storage_operation_run(
history_id,
run["run_id"],
include_items_on_terminal=True,
@@ -857,7 +862,7 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase):
baseline_default_count, baseline_separate_count = self._store_file_counts()
first_execute = self._execute_snapshot(history_id, preview["snapshot_id"], skip_ineligible=True)
first_run = self.dataset_populator.wait_for_storage_run(
first_run = self.dataset_populator.wait_for_bulk_storage_operation_run(
history_id,
first_execute["run"]["run_id"],
include_items_on_terminal=True,
@@ -889,7 +894,7 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase):
self._assert_dataset_store_and_content(history_id, control["id"], DEFAULT_OBJECT_STORE_ID, "control\n")
second_execute = self._execute_snapshot(history_id, preview["snapshot_id"], skip_ineligible=True)
second_run = self.dataset_populator.wait_for_storage_run(
second_run = self.dataset_populator.wait_for_bulk_storage_operation_run(
history_id,
second_execute["run"]["run_id"],
include_items_on_terminal=True,
@@ -947,7 +952,7 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase):
skip_ineligible=False,
force_checksum_mismatch=True,
)
first_run_status = self.dataset_populator.storage_run_status(history_id, first_run_id)
first_run_status = self.dataset_populator.bulk_storage_operation_run_status(history_id, first_run_id)
self._assert_run_counts(first_run_status["run"], succeeded=0, failed=1, skipped=0)
assert first_run_status["run"]["total_bytes_processed"] == 0
@@ -972,7 +977,7 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase):
# Re-run same snapshot without forced corruption; move should now succeed.
second_run_id = self._execute_snapshot_sync(sa_session, snapshot, skip_ineligible=False)
second_run_status = self.dataset_populator.storage_run_status(history_id, second_run_id)
second_run_status = self.dataset_populator.bulk_storage_operation_run_status(history_id, second_run_id)
self._assert_run_counts(second_run_status["run"], succeeded=1, failed=0, skipped=0)
assert second_run_status["run"]["total_bytes_processed"] > 0
@@ -1099,7 +1104,7 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase):
assert result.state == StorageOperationRunState.completed
encoded_run_id = self._app.security.encode_id(recovered_run.id)
final = self.dataset_populator.storage_run_status(history_id, encoded_run_id)
final = self.dataset_populator.bulk_storage_operation_run_status(history_id, encoded_run_id)
self._assert_run_counts(final["run"], succeeded=2, failed=0, skipped=0)
items = self._run_items(history_id, encoded_run_id)
@@ -1213,7 +1218,7 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase):
assert result.state == StorageOperationRunState.completed
encoded_run_id = self._app.security.encode_id(recovered_run.id)
final = self.dataset_populator.storage_run_status(history_id, encoded_run_id)
final = self.dataset_populator.bulk_storage_operation_run_status(history_id, encoded_run_id)
self._assert_run_counts(final["run"], succeeded=1, failed=0, skipped=0)
items = self._run_items(history_id, encoded_run_id)