opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-24 16:30:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #14989 from davelopez/fix_import_history_using_tasks

Browse Source 
Fix discarded datasets when importing history from file sources using tasks
This commit is contained in:
Björn Grüning
2022-11-17 22:34:36 +01:00
committed by GitHub
parent 3d58309677 eed791167e
commit ea43465436
7 changed files with 115 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/managers/model_stores.py
+1 -1
View File
@@ -168,7 +168,6 @@ class ModelStoreManager:
def import_model_store(self, request: ImportModelStoreTaskRequest):
import_options = ImportOptions(
discarded_data=ImportDiscardedDataType.FORCE,
allow_library_creation=request.for_library,
)
history_id = request.history_id
@@ -186,6 +185,7 @@ class ModelStoreManager:
self._app,
galaxy_user,
import_options,
model_store_format=request.model_store_format,
)
new_history = history is None and not request.for_library
if new_history:
lib/galaxy/model/store/__init__.py
+24 -9
View File
@@ -1,6 +1,7 @@
import abc
import contextlib
import datetime
import logging
import os
import shutil
import tarfile
@@ -54,6 +55,7 @@ from galaxy.model.orm.util import (
add_object_to_session,
get_object_session,
)
from galaxy.model.tags import GalaxyTagHandler
from galaxy.objectstore import ObjectStore
from galaxy.schema.bco import (
BioComputeObjectCore,
@@ -78,6 +80,7 @@ from galaxy.schema.bco.util import (
get_contributors,
write_to_file,
)
from galaxy.schema.schema import ModelStoreFormat
from galaxy.security.idencoding import IdEncodingHelper
from galaxy.util import (
FILENAME_VALID_CHARS,
@@ -101,6 +104,7 @@ from ... import model
if TYPE_CHECKING:
from galaxy.managers.workflows import WorkflowContentsManager
log = logging.getLogger(__name__)
ObjectKeyType = Union[str, int]
@@ -134,6 +138,7 @@ class StoreAppProtocol(Protocol):
datatypes_registry: Registry
object_store: ObjectStore
security: IdEncodingHelper
tag_handler: GalaxyTagHandler
model: GalaxyModelMapping
file_sources: ConfiguredFileSources
workflow_contents_manager: "WorkflowContentsManager"
@@ -623,9 +628,18 @@ class ModelImportStore(metaclass=abc.ABCMeta):
else:
# Need a user to run library jobs to generate metadata...
pass
self.app.datatypes_registry.set_external_metadata_tool.regenerate_imported_metadata_if_needed(
dataset_instance, history, **regenerate_kwds
)
if self.app.datatypes_registry.set_external_metadata_tool:
self.app.datatypes_registry.set_external_metadata_tool.regenerate_imported_metadata_if_needed(
dataset_instance, history, **regenerate_kwds
)
else:
# Try to set metadata directly. @mvdbeek thinks we should only record the datasets
try:
if dataset_instance.has_metadata_files:
dataset_instance.datatype.set_meta(dataset_instance)
except Exception:
log.debug(f"Metadata setting failed on {dataset_instance}", exc_info=True)
dataset_instance.dataset.state = dataset_instance.dataset.states.FAILED_METADATA
if model_class == "HistoryDatasetAssociation":
if object_key in dataset_attrs:
@@ -2646,7 +2660,7 @@ def source_to_import_store(
app: StoreAppProtocol,
galaxy_user: Optional[model.User],
import_options: Optional[ImportOptions],
model_store_format: Optional[str] = None,
model_store_format: Optional[ModelStoreFormat] = None,
) -> ModelImportStore:
if isinstance(source, dict):
if model_store_format is not None:
@@ -2662,6 +2676,7 @@ def source_to_import_store(
else:
source_uri: str = str(source)
delete = False
tag_handler = app.tag_handler.create_tag_handler_session()
if source_uri.startswith("file://"):
source_uri = source_uri[len("file://") :]
if "://" in source_uri:
@@ -2680,11 +2695,11 @@ def source_to_import_store(
)
elif os.path.isdir(target_path):
model_import_store = get_import_model_store_for_directory(
target_path, import_options=import_options, app=app, user=galaxy_user
target_path, import_options=import_options, app=app, user=galaxy_user, tag_handler=tag_handler
)
else:
model_store_format = model_store_format or "tgz"
if model_store_format in ["tar.gz", "tgz", "tar"]:
model_store_format = model_store_format or ModelStoreFormat.TGZ
if ModelStoreFormat.is_compressed(model_store_format):
try:
temp_dir = mkdtemp()
target_dir = CompressedFile(target_path).extract(temp_dir)
@@ -2692,9 +2707,9 @@ def source_to_import_store(
if delete:
os.remove(target_path)
model_import_store = get_import_model_store_for_directory(
target_dir, import_options=import_options, app=app, user=galaxy_user
target_dir, import_options=import_options, app=app, user=galaxy_user, tag_handler=tag_handler
)
elif model_store_format in ["bag.gz", "bag.tar", "bag.zip"]:
elif ModelStoreFormat.is_bag(model_store_format):
model_import_store = BagArchiveImportModelStore(
target_path, import_options=import_options, app=app, user=galaxy_user
)
lib/galaxy/schema/schema.py
+8
View File
@@ -1305,6 +1305,14 @@ class ModelStoreFormat(str, Enum):
ROCRATE_ZIP = "rocrate.zip"
BCO_JSON = "bco.json"
@classmethod
def is_compressed(cls, value: "ModelStoreFormat"):
return value in [cls.TAR_DOT_GZ, cls.TGZ, cls.TAR, cls.ROCRATE_ZIP]
@classmethod
def is_bag(cls, value: "ModelStoreFormat"):
return value in [cls.BAG_DOT_TAR, cls.BAG_DOT_TGZ, cls.BAG_DOT_ZIP]
class StoreContentSource(Model):
store_content_uri: Optional[str]
lib/galaxy/schema/tasks.py
+2
View File
@@ -10,6 +10,7 @@ from .schema import (
BcoGenerationParametersMixin,
DatasetSourceType,
HistoryContentType,
ModelStoreFormat,
StoreExportPayload,
WriteStoreToPayload,
)
@@ -88,6 +89,7 @@ class ImportModelStoreTaskRequest(BaseModel):
history_id: Optional[int]
source_uri: str
for_library: bool
model_store_format: Optional[ModelStoreFormat]
class MaterializeDatasetInstanceTaskRequest(BaseModel):
lib/galaxy/webapps/galaxy/services/histories.py
+1
View File
@@ -281,6 +281,7 @@ class HistoriesService(ServiceBase, ConsumesModelStores, ServesExportStores):
user=trans.async_request_user,
source_uri=source_uri,
for_library=False,
model_store_format=payload.model_store_format,
)
result = import_model_store.delay(request=request)
return async_task_summary(result)
lib/galaxy_test/base/populators.py
+33 -4
View File
@@ -477,17 +477,23 @@ class BaseDatasetPopulator(BasePopulator):
return create_response
def create_from_store(
self, store_dict: Optional[Dict[str, Any]] = None, store_path: Optional[str] = None
self,
store_dict: Optional[Dict[str, Any]] = None,
store_path: Optional[str] = None,
model_store_format: Optional[str] = None,
) -> Dict[str, Any]:
payload = _store_payload(store_dict=store_dict, store_path=store_path)
payload = _store_payload(store_dict=store_dict, store_path=store_path, model_store_format=model_store_format)
create_response = self.create_from_store_raw(payload)
api_asserts.assert_status_code_is_ok(create_response)
return create_response.json()
def create_from_store_async(
self, store_dict: Optional[Dict[str, Any]] = None, store_path: Optional[str] = None
self,
store_dict: Optional[Dict[str, Any]] = None,
store_path: Optional[str] = None,
model_store_format: Optional[str] = None,
) -> Dict[str, Any]:
payload = _store_payload(store_dict=store_dict, store_path=store_path)
payload = _store_payload(store_dict=store_dict, store_path=store_path, model_store_format=model_store_format)
create_response = self.create_from_store_raw_async(payload)
create_response.raise_for_status()
return create_response.json()
@@ -1223,7 +1229,9 @@ class BaseDatasetPopulator(BasePopulator):
def wait_on_task(self, async_task_response: Response):
task_id = async_task_response.json()["id"]
return self.wait_on_task_id(task_id)
def wait_on_task_id(self, task_id: str):
def state():
state_response = self._get(f"tasks/{task_id}/state")
state_response.raise_for_status()
@@ -1324,6 +1332,27 @@ class BaseDatasetPopulator(BasePopulator):
)
return request
def export_history_to_uri_async(
self, history_id: str, target_uri: str, model_store_format: str = "tgz", include_files: bool = True
):
url = f"histories/{history_id}/write_store"
download_response = self._post(
url,
dict(target_uri=target_uri, include_files=include_files, model_store_format=model_store_format),
json=True,
)
api_asserts.assert_status_code_is_ok(download_response)
task_ok = self.wait_on_task(download_response)
assert task_ok, f"Task: Writing history to {target_uri} task failed"
def import_history_from_uri_async(self, target_uri: str, model_store_format: str):
import_async_response = self.create_from_store_async(
store_path=target_uri, model_store_format=model_store_format
)
task_id = import_async_response["id"]
task_ok = self.wait_on_task_id(task_id)
assert task_ok, f"Task: Import history from {target_uri} failed"
class GalaxyInteractorHttpMixin:
galaxy_interactor: ApiTestInteractor
test/integration/test_history_import_export.py
+46 -1
View File
@@ -1,4 +1,5 @@
import tarfile
from uuid import uuid4
from galaxy.model.unittest_utils.store_fixtures import (
deferred_hda_model_store_dict,
@@ -12,6 +13,7 @@ from galaxy_test.base.populators import (
DatasetCollectionPopulator,
DatasetPopulator,
)
from galaxy_test.driver.integration_setup import PosixFileSourceSetup
from galaxy_test.driver.integration_util import IntegrationTestCase
@@ -29,10 +31,17 @@ class TestImportExportHistoryOutputsToWorkingDirIntegration(ImportExportTests, I
self._set_up_populators()
class TestImportExportHistoryViaTasksIntegration(ImportExportTests, IntegrationTestCase, UsesCeleryTasks):
class TestImportExportHistoryViaTasksIntegration(
ImportExportTests, IntegrationTestCase, UsesCeleryTasks, PosixFileSourceSetup
):
task_based = True
framework_tool_and_types = True
@classmethod
def handle_galaxy_config_kwds(cls, config):
PosixFileSourceSetup.handle_galaxy_config_kwds(config, cls)
UsesCeleryTasks.handle_galaxy_config_kwds(config)
def setUp(self):
super().setUp()
self._set_up_populators()
@@ -48,6 +57,42 @@ class TestImportExportHistoryViaTasksIntegration(ImportExportTests, IntegrationT
"task based import history",
)
def test_import_model_store_from_file_source_async_with_format(self):
history_name = f"for_export_format_async_{uuid4()}"
history_id = self.dataset_populator.setup_history_for_export_testing(history_name)
# Add bam dataset to test metadata generation on import
self.dataset_populator.new_dataset(
history_id, content=open(self.test_data_resolver.get_filename("1.bam"), "rb"), file_type="bam", wait=True
)
model_store_format = "rocrate.zip"
target_uri = f"gxfiles://posix_test/history.{model_store_format}"
self.dataset_populator.export_history_to_uri_async(history_id, target_uri, model_store_format)
self.dataset_populator.import_history_from_uri_async(target_uri, model_store_format)
last_history = self._get("histories?limit=1").json()
assert len(last_history) == 1
imported_history = last_history[0]
imported_history_id = imported_history["id"]
assert imported_history_id != history_id
assert imported_history["name"] == history_name
self.dataset_populator.wait_for_history(imported_history_id)
history_contents = self.dataset_populator.get_history_contents(imported_history_id)
assert len(history_contents) == 3
# Only deleted datasets should appear as "discarded"
for dataset in history_contents:
if dataset["deleted"] is True:
assert dataset["state"] == "discarded"
else:
assert dataset["state"] == "ok"
# Check metadata generation
if dataset["extension"] == "bam":
imported_bam_details = self.dataset_populator.get_history_dataset_details(
imported_history_id, dataset_id=dataset["id"]
)
bai_metadata = imported_bam_details["meta_files"][0]
assert bai_metadata["file_type"] == "bam_index"
class TestImportExportHistoryContentsViaTasksIntegration(IntegrationTestCase, UsesCeleryTasks):
dataset_populator: DatasetPopulator