mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-10-25 00:12:08 +08:00
Merge pull request #13348 from mvdbeek/upload_metadata
Fix for uploading metadata to remote object store
This commit is contained in:
@@ -432,6 +432,7 @@ def set_metadata_portable():
|
||||
) # setting metadata has failed somehow
|
||||
|
||||
if export_store:
|
||||
export_store.push_metadata_files()
|
||||
export_store._finalize()
|
||||
write_job_metadata(tool_job_working_directory, job_metadata, set_meta, tool_provided_metadata)
|
||||
|
||||
|
||||
@@ -7931,15 +7931,31 @@ class MetadataFile(Base, StorableObject, Serializable):
|
||||
self.library_dataset = dataset
|
||||
self.name = name
|
||||
|
||||
@property
|
||||
def dataset(self) -> Optional[Dataset]:
|
||||
da = self.history_dataset or self.library_dataset
|
||||
return da and da.dataset
|
||||
|
||||
def update_from_file(self, file_name):
|
||||
if not self.dataset:
|
||||
raise Exception("Attempted to write MetadataFile, but no DatasetAssociation set")
|
||||
self.dataset.object_store.update_from_file(
|
||||
self,
|
||||
file_name=file_name,
|
||||
extra_dir="_metadata_files",
|
||||
extra_dir_at_root=True,
|
||||
alt_name=os.path.basename(self.file_name),
|
||||
)
|
||||
|
||||
@property
|
||||
def file_name(self):
|
||||
# Ensure the directory structure and the metadata file object exist
|
||||
try:
|
||||
da = self.history_dataset or self.library_dataset
|
||||
if self.object_store_id is None and da is not None:
|
||||
self.object_store_id = da.dataset.object_store_id
|
||||
object_store = da.dataset.object_store
|
||||
store_by = object_store.get_store_by(da.dataset)
|
||||
dataset = self.dataset
|
||||
if self.object_store_id is None and dataset is not None:
|
||||
self.object_store_id = dataset.object_store_id
|
||||
object_store = dataset.object_store
|
||||
store_by = object_store.get_store_by(dataset)
|
||||
if store_by == "id" and self.id is None:
|
||||
self.flush()
|
||||
identifier = getattr(self, store_by)
|
||||
|
||||
@@ -6,7 +6,6 @@ import copy
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
import weakref
|
||||
@@ -604,6 +603,10 @@ class FileParameter(MetadataParameter):
|
||||
session = target_context._object_session(target_context.parent)
|
||||
value = self.wrap(value, session=session)
|
||||
target_dataset = target_context.parent.dataset
|
||||
if value and not value.id:
|
||||
# This is a new MetadataFile object, we're not copying to another dataset.
|
||||
# Just use it.
|
||||
return self.unwrap(value)
|
||||
if value and target_dataset.object_store.exists(target_dataset):
|
||||
# Only copy MetadataFile if the target dataset has been created in an object store.
|
||||
# All current datatypes re-generate MetadataFile objects when setting metadata,
|
||||
@@ -611,10 +614,10 @@ class FileParameter(MetadataParameter):
|
||||
new_value = galaxy.model.MetadataFile(dataset=target_context.parent, name=self.spec.name)
|
||||
session.add(new_value)
|
||||
try:
|
||||
shutil.copy(value.file_name, new_value.file_name)
|
||||
new_value.update_from_file(value.file_name)
|
||||
except AssertionError:
|
||||
session(target_context.parent).flush()
|
||||
shutil.copy(value.file_name, new_value.file_name)
|
||||
new_value.update_from_file(value.file_name)
|
||||
return self.unwrap(new_value)
|
||||
return None
|
||||
|
||||
@@ -645,13 +648,7 @@ class FileParameter(MetadataParameter):
|
||||
# Job may have run with a different (non-local) tmp/working
|
||||
# directory. Correct.
|
||||
file_name = path_rewriter(file_name)
|
||||
parent.dataset.object_store.update_from_file(
|
||||
mf,
|
||||
file_name=file_name,
|
||||
extra_dir="_metadata_files",
|
||||
extra_dir_at_root=True,
|
||||
alt_name=os.path.basename(mf.file_name),
|
||||
)
|
||||
mf.update_from_file(file_name)
|
||||
os.unlink(file_name)
|
||||
value = mf.id
|
||||
return value
|
||||
|
||||
@@ -87,10 +87,12 @@ class SessionlessContext:
|
||||
return Bunch(find=find, get=find, filter_by=filter_by)
|
||||
|
||||
|
||||
def replace_metadata_file(metadata: Dict[str, Any], dataset_instance: model.DatasetInstance):
|
||||
def replace_metadata_file(metadata: Dict[str, Any], dataset_instance: model.DatasetInstance, sa_session):
|
||||
def remap_objects(p, k, obj):
|
||||
if isinstance(obj, dict) and "model_class" in obj and obj["model_class"] == "MetadataFile":
|
||||
return (k, model.MetadataFile(dataset=dataset_instance, uuid=obj["uuid"]))
|
||||
metadata_file = model.MetadataFile(dataset=dataset_instance, uuid=obj["uuid"])
|
||||
sa_session.add(metadata_file)
|
||||
return (k, metadata_file)
|
||||
return (k, obj)
|
||||
|
||||
return remap(metadata, remap_objects)
|
||||
@@ -293,7 +295,7 @@ class ModelImportStore(metaclass=abc.ABCMeta):
|
||||
if attribute in dataset_attrs:
|
||||
value = dataset_attrs.get(attribute)
|
||||
if attribute == "metadata":
|
||||
value = replace_metadata_file(value, dataset_instance)
|
||||
value = replace_metadata_file(value, dataset_instance, self.sa_session)
|
||||
setattr(dataset_instance, attribute, value)
|
||||
|
||||
handle_dataset_object_edit(dataset_instance)
|
||||
@@ -345,7 +347,7 @@ class ModelImportStore(metaclass=abc.ABCMeta):
|
||||
dataset_instance._metadata_collection = MetadataCollection(
|
||||
dataset_instance, session=self.sa_session
|
||||
)
|
||||
metadata = replace_metadata_file(metadata, dataset_instance)
|
||||
metadata = replace_metadata_file(metadata, dataset_instance, self.sa_session)
|
||||
dataset_instance._metadata = metadata
|
||||
self._attach_raw_id_if_editing(dataset_instance, dataset_attrs)
|
||||
|
||||
@@ -1296,6 +1298,12 @@ class DirectoryModelExportStore(ModelExportStore):
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def push_metadata_files(self):
|
||||
for dataset in self.included_datasets:
|
||||
for metadata_element in dataset.metadata.values():
|
||||
if isinstance(metadata_element, model.MetadataFile):
|
||||
metadata_element.update_from_file(metadata_element.file_name)
|
||||
|
||||
def export_job(self, job: model.Job, tool=None, include_job_data=True):
|
||||
self.export_jobs([job], include_job_data=include_job_data)
|
||||
tool_source = getattr(tool, "tool_source", None)
|
||||
|
||||
@@ -256,8 +256,8 @@ class FastAPIDatasets:
|
||||
trans=DependsOnTrans,
|
||||
history_id: EncodedDatabaseIdField = HistoryIDPathParam,
|
||||
history_content_id: EncodedDatabaseIdField = DatasetIDPathParam,
|
||||
metadata_file: Optional[str] = Query(
|
||||
default=None,
|
||||
metadata_file: str = Query(
|
||||
...,
|
||||
description="The name of the metadata file to retrieve.",
|
||||
),
|
||||
):
|
||||
@@ -446,7 +446,7 @@ class DatasetsController(BaseGalaxyAPIController):
|
||||
return self.service.get_content_as_text(trans, dataset_id)
|
||||
|
||||
@web.expose_api_raw_anonymous_and_sessionless
|
||||
def get_metadata_file(self, trans, history_content_id, history_id, metadata_file=None, **kwd):
|
||||
def get_metadata_file(self, trans, history_content_id, history_id, metadata_file, **kwd):
|
||||
"""
|
||||
GET /api/histories/{history_id}/contents/{history_content_id}/metadata_file
|
||||
"""
|
||||
|
||||
@@ -135,7 +135,7 @@ class DatasetInterface(BaseUIController, UsesAnnotations, UsesItemRatings, UsesE
|
||||
return "This link may not be followed from within Galaxy."
|
||||
|
||||
@web.expose
|
||||
def get_metadata_file(self, trans, hda_id, metadata_name):
|
||||
def get_metadata_file(self, trans, hda_id, metadata_file):
|
||||
"""Allows the downloading of metadata files associated with datasets (eg. bai index for bam files)"""
|
||||
data = trans.sa_session.query(trans.app.model.HistoryDatasetAssociation).get(self.decode_id(hda_id))
|
||||
if not data or not self._can_access_dataset(trans, data):
|
||||
@@ -143,10 +143,10 @@ class DatasetInterface(BaseUIController, UsesAnnotations, UsesItemRatings, UsesE
|
||||
|
||||
fname = "".join(c in util.FILENAME_VALID_CHARS and c or "_" for c in data.name)[0:150]
|
||||
|
||||
file_ext = data.metadata.spec.get(metadata_name).get("file_ext", metadata_name)
|
||||
file_ext = data.metadata.spec.get(metadata_file).get("file_ext", metadata_file)
|
||||
trans.response.headers["Content-Type"] = "application/octet-stream"
|
||||
trans.response.headers["Content-Disposition"] = f'attachment; filename="Galaxy{data.hid}-[{fname}].{file_ext}"'
|
||||
return open(data.metadata.get(metadata_name).file_name, "rb")
|
||||
return open(data.metadata.get(metadata_file).file_name, "rb")
|
||||
|
||||
def _check_dataset(self, trans, hda_id):
|
||||
# DEPRECATION: We still support unencoded ids for backward compatibility
|
||||
|
||||
@@ -434,7 +434,7 @@ class DatasetsService(ServiceBase, UsesVisualizationMixin):
|
||||
self,
|
||||
trans: ProvidesHistoryContext,
|
||||
history_content_id: EncodedDatabaseIdField,
|
||||
metadata_file: Optional[str] = None,
|
||||
metadata_file: str,
|
||||
open_file: bool = False,
|
||||
):
|
||||
"""
|
||||
|
||||
@@ -20,6 +20,13 @@ class CacheOperationTestCase(BaseSwiftObjectStoreIntegrationTestCase):
|
||||
hda = self.dataset_populator.new_dataset(history_id, content="123", wait=True)
|
||||
return hda
|
||||
|
||||
def upload_bam_dataset(self):
|
||||
history_id = self.dataset_populator.new_history()
|
||||
hda = self.dataset_populator.new_dataset(
|
||||
history_id, content=open(self.test_data_resolver.get_filename("1.bam"), "rb"), file_type="bam", wait=True
|
||||
)
|
||||
return hda
|
||||
|
||||
def test_cache_populated(self):
|
||||
self.upload_dataset()
|
||||
assert files_count(self.object_store_cache_path) == 1
|
||||
@@ -34,6 +41,23 @@ class CacheOperationTestCase(BaseSwiftObjectStoreIntegrationTestCase):
|
||||
assert content == "123\n"
|
||||
assert files_count(self.object_store_cache_path) == 1
|
||||
|
||||
def test_cache_repopulated_metadata_file(self):
|
||||
hda = self.upload_bam_dataset()
|
||||
assert files_count(self.object_store_cache_path) == 2
|
||||
response = self._get(
|
||||
f'histories/{hda["history_id"]}/contents/{hda["id"]}/metadata_file?metadata_file=bam_index'
|
||||
)
|
||||
assert len(response.content) > 0
|
||||
response.raise_for_status()
|
||||
shutil.rmtree(self.object_store_cache_path)
|
||||
assert files_count(self.object_store_cache_path) == 0
|
||||
response = self._get(
|
||||
f'histories/{hda["history_id"]}/contents/{hda["id"]}/metadata_file?metadata_file=bam_index'
|
||||
)
|
||||
response.raise_for_status()
|
||||
assert files_count(self.object_store_cache_path) == 1
|
||||
assert len(response.content) > 0
|
||||
|
||||
def test_delete_item_not_in_cache(self):
|
||||
hda = self.upload_dataset()
|
||||
assert files_count(self.object_store_cache_path) == 1
|
||||
|
||||
Reference in New Issue
Block a user