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

Merge pull request #11816 from mvdbeek/performance_fix_setup_external_metadata

Browse Source 
[21.01] Backport #11815: Don't write job and job-related attributes when creating export model
This commit is contained in:
Björn Grüning
2021-04-12 15:17:11 +02:00
committed by GitHub
parent 08694fd05f 897590b2f3
commit 9348fdae7f
3 changed files with 142 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/metadata/__init__.py
+1 -1
View File
@@ -157,7 +157,7 @@ class PortableDirectoryMetadataGenerator(MetadataCollectionStrategy):
# export model objects and object store configuration for extended metadata also.
export_directory = os.path.join(metadata_dir, "outputs_new")
with DirectoryModelExportStore(export_directory, for_edit=True, serialize_dataset_objects=True) as export_store:
with DirectoryModelExportStore(export_directory, for_edit=True, serialize_dataset_objects=True, serialize_jobs=False) as export_store:
for dataset in datasets_dict.values():
export_store.add_dataset(dataset)
lib/galaxy/metadata/set_metadata.py
+1 -1
View File
@@ -188,7 +188,7 @@ def set_metadata_portable():
job_context = ExpressionContext(dict(stdout=tool_stdout, stderr=tool_stderr))
# Load outputs.
export_store = store.DirectoryModelExportStore('metadata/outputs_populated', serialize_dataset_objects=True, for_edit=True, strip_metadata_files=False)
export_store = store.DirectoryModelExportStore('metadata/outputs_populated', serialize_dataset_objects=True, for_edit=True, strip_metadata_files=False, serialize_jobs=False)
try:
import_model_store = store.imported_store_for_metadata('metadata/outputs_new', object_store=object_store)
except AssertionError:
lib/galaxy/model/store/__init__.py
+140 -137
View File
@@ -819,16 +819,17 @@ class BaseDirectoryImportModelStore(ModelImportStore):
def jobs_properties(self):
jobs_attr_file_name = os.path.join(self.archive_dir, ATTRS_FILENAME_JOBS)
jobs_attrs = load(open(jobs_attr_file_name))
return jobs_attrs
try:
return load(open(jobs_attr_file_name))
except FileNotFoundError:
return []
def implicit_collection_jobs_properties(self):
implicit_collection_jobs_attrs_file_name = os.path.join(self.archive_dir, ATTRS_FILENAME_IMPLICIT_COLLECTION_JOBS)
if os.path.exists(implicit_collection_jobs_attrs_file_name):
implicit_collection_jobs_attrs = load(open(implicit_collection_jobs_attrs_file_name))
else:
implicit_collection_jobs_attrs = []
return implicit_collection_jobs_attrs
try:
return load(open(implicit_collection_jobs_attrs_file_name))
except FileNotFoundError:
return []
class DirectoryImportModelStore1901(BaseDirectoryImportModelStore):
@@ -1010,7 +1011,7 @@ class ModelExportStore(metaclass=abc.ABCMeta):
class DirectoryModelExportStore(ModelExportStore):
def __init__(self, export_directory, app=None, for_edit=False, serialize_dataset_objects=None, export_files=None, strip_metadata_files=True):
def __init__(self, export_directory, app=None, for_edit=False, serialize_dataset_objects=None, export_files=None, strip_metadata_files=True, serialize_jobs=True):
"""
:param export_directory: path to export directory. Will be created if it does not exist.
:param app: Galaxy App or app-like object. Must be provided if `for_edit` and/or `serialize_dataset_objects` are True
@@ -1018,6 +1019,7 @@ class DirectoryModelExportStore(ModelExportStore):
:param serialize_dataset_objects: If True will encode IDs using the host secret. Defaults `for_edit`.
:param export_files: How files should be exported, can be 'symlink', 'copy' or None, in which case files
will not be serialized.
:param serialize_jobs: Include job data in model export. Not needed for set_metadata script.
"""
if not os.path.exists(export_directory):
os.makedirs(export_directory)
@@ -1030,6 +1032,7 @@ class DirectoryModelExportStore(ModelExportStore):
sessionless = True
security = IdEncodingHelper(id_secret="randomdoesntmatter")
self.serialize_jobs = serialize_jobs
self.sessionless = sessionless
self.security = security
@@ -1247,126 +1250,7 @@ class DirectoryModelExportStore(ModelExportStore):
with open(collections_attrs_filename, 'w') as collections_attrs_out:
collections_attrs_out.write(to_json(self.collections_attrs))
#
# Write jobs attributes file.
#
# Get all jobs associated with included HDAs.
jobs_dict = {}
implicit_collection_jobs_dict = {}
def record_associated_jobs(obj):
# Get the job object.
job = None
for assoc in getattr(obj, 'creating_job_associations', []):
# For mapped over jobs obj could be DatasetCollection, which has no creating_job_association
job = assoc.job
break
if not job:
# No viable job.
return
jobs_dict[job.id] = job
icja = job.implicit_collection_jobs_association
if icja:
implicit_collection_jobs = icja.implicit_collection_jobs
implicit_collection_jobs_dict[implicit_collection_jobs.id] = implicit_collection_jobs
for hda, _include_files in self.included_datasets.values():
# Get the associated job, if any. If this hda was copied from another,
# we need to find the job that created the origial hda
job_hda = hda
while job_hda.copied_from_history_dataset_association: # should this check library datasets as well?
job_hda = job_hda.copied_from_history_dataset_association
if not job_hda.creating_job_associations:
# No viable HDA found.
continue
record_associated_jobs(job_hda)
for hdca in self.included_collections:
record_associated_jobs(hdca)
# Get jobs' attributes.
jobs_attrs = []
for job in jobs_dict.values():
# Don't attempt to serialize jobs for editing... yet at least.
if self.serialization_options.for_edit:
continue
job_attrs = job.serialize(self.security, self.serialization_options)
# -- Get input, output datasets. --
input_dataset_mapping = {}
output_dataset_mapping = {}
input_dataset_collection_mapping = {}
input_dataset_collection_element_mapping = {}
output_dataset_collection_mapping = {}
implicit_output_dataset_collection_mapping = {}
for assoc in job.input_datasets:
# Optional data inputs will not have a dataset.
if assoc.dataset:
name = assoc.name
if name not in input_dataset_mapping:
input_dataset_mapping[name] = []
input_dataset_mapping[name].append(self.exported_key(assoc.dataset))
for assoc in job.output_datasets:
# Optional data inputs will not have a dataset.
if assoc.dataset:
name = assoc.name
if name not in output_dataset_mapping:
output_dataset_mapping[name] = []
output_dataset_mapping[name].append(self.exported_key(assoc.dataset))
for assoc in job.input_dataset_collections:
# Optional data inputs will not have a dataset.
if assoc.dataset_collection:
name = assoc.name
if name not in input_dataset_collection_mapping:
input_dataset_collection_mapping[name] = []
input_dataset_collection_mapping[name].append(self.exported_key(assoc.dataset_collection))
for assoc in job.input_dataset_collection_elements:
if assoc.dataset_collection_element:
name = assoc.name
if name not in input_dataset_collection_element_mapping:
input_dataset_collection_element_mapping[name] = []
input_dataset_collection_element_mapping[name].append(self.exported_key(assoc.dataset_collection_element))
for assoc in job.output_dataset_collection_instances:
# Optional data outputs will not have a dataset.
if assoc.dataset_collection_instance:
name = assoc.name
if name not in output_dataset_collection_mapping:
output_dataset_collection_mapping[name] = []
output_dataset_collection_mapping[name].append(self.exported_key(assoc.dataset_collection_instance))
for assoc in job.output_dataset_collections:
if assoc.dataset_collection:
name = assoc.name
if name not in implicit_output_dataset_collection_mapping:
implicit_output_dataset_collection_mapping[name] = []
implicit_output_dataset_collection_mapping[name].append(self.exported_key(assoc.dataset_collection))
job_attrs['input_dataset_mapping'] = input_dataset_mapping
job_attrs['input_dataset_collection_mapping'] = input_dataset_collection_mapping
job_attrs['input_dataset_collection_element_mapping'] = input_dataset_collection_element_mapping
job_attrs['output_dataset_mapping'] = output_dataset_mapping
job_attrs['output_dataset_collection_mapping'] = output_dataset_collection_mapping
job_attrs['implicit_output_dataset_collection_mapping'] = implicit_output_dataset_collection_mapping
jobs_attrs.append(job_attrs)
for job_id, job_output_dataset_associations in self.job_output_dataset_associations.items():
output_dataset_mapping = {}
for name, dataset in job_output_dataset_associations.items():
@@ -1375,23 +1259,142 @@ class DirectoryModelExportStore(ModelExportStore):
output_dataset_mapping[name].append(self.exported_key(dataset))
jobs_attrs.append({"id": job_id, 'output_dataset_mapping': output_dataset_mapping})
icjs_attrs = []
for icj in implicit_collection_jobs_dict.values():
icj_attrs = icj.serialize(self.security, self.serialization_options)
icjs_attrs.append(icj_attrs)
if self.serialize_jobs:
jobs_attrs_filename = os.path.join(export_directory, ATTRS_FILENAME_JOBS)
with open(jobs_attrs_filename, 'w') as jobs_attrs_out:
jobs_attrs_out.write(json_encoder.encode(jobs_attrs))
#
# Write jobs attributes file.
#
icjs_attrs_filename = os.path.join(export_directory, ATTRS_FILENAME_IMPLICIT_COLLECTION_JOBS)
with open(icjs_attrs_filename, 'w') as icjs_attrs_out:
icjs_attrs_out.write(json_encoder.encode(icjs_attrs))
# Get all jobs associated with included HDAs.
jobs_dict = {}
implicit_collection_jobs_dict = {}
def record_associated_jobs(obj):
# Get the job object.
job = None
for assoc in getattr(obj, 'creating_job_associations', []):
# For mapped over jobs obj could be DatasetCollection, which has no creating_job_association
job = assoc.job
break
if not job:
# No viable job.
return
jobs_dict[job.id] = job
icja = job.implicit_collection_jobs_association
if icja:
implicit_collection_jobs = icja.implicit_collection_jobs
implicit_collection_jobs_dict[implicit_collection_jobs.id] = implicit_collection_jobs
for hda, _include_files in self.included_datasets.values():
# Get the associated job, if any. If this hda was copied from another,
# we need to find the job that created the origial hda
job_hda = hda
while job_hda.copied_from_history_dataset_association: # should this check library datasets as well?
job_hda = job_hda.copied_from_history_dataset_association
if not job_hda.creating_job_associations:
# No viable HDA found.
continue
record_associated_jobs(job_hda)
for hdca in self.included_collections:
record_associated_jobs(hdca)
# Get jobs' attributes.
for job in jobs_dict.values():
if self.serialization_options.for_edit:
continue
job_attrs = job.serialize(self.security, self.serialization_options)
# -- Get input, output datasets. --
input_dataset_mapping = {}
output_dataset_mapping = {}
input_dataset_collection_mapping = {}
input_dataset_collection_element_mapping = {}
output_dataset_collection_mapping = {}
implicit_output_dataset_collection_mapping = {}
for assoc in job.input_datasets:
# Optional data inputs will not have a dataset.
if assoc.dataset:
name = assoc.name
if name not in input_dataset_mapping:
input_dataset_mapping[name] = []
input_dataset_mapping[name].append(self.exported_key(assoc.dataset))
for assoc in job.output_datasets:
# Optional data inputs will not have a dataset.
if assoc.dataset:
name = assoc.name
if name not in output_dataset_mapping:
output_dataset_mapping[name] = []
output_dataset_mapping[name].append(self.exported_key(assoc.dataset))
for assoc in job.input_dataset_collections:
# Optional data inputs will not have a dataset.
if assoc.dataset_collection:
name = assoc.name
if name not in input_dataset_collection_mapping:
input_dataset_collection_mapping[name] = []
input_dataset_collection_mapping[name].append(self.exported_key(assoc.dataset_collection))
for assoc in job.input_dataset_collection_elements:
if assoc.dataset_collection_element:
name = assoc.name
if name not in input_dataset_collection_element_mapping:
input_dataset_collection_element_mapping[name] = []
input_dataset_collection_element_mapping[name].append(self.exported_key(assoc.dataset_collection_element))
for assoc in job.output_dataset_collection_instances:
# Optional data outputs will not have a dataset.
if assoc.dataset_collection_instance:
name = assoc.name
if name not in output_dataset_collection_mapping:
output_dataset_collection_mapping[name] = []
output_dataset_collection_mapping[name].append(self.exported_key(assoc.dataset_collection_instance))
for assoc in job.output_dataset_collections:
if assoc.dataset_collection:
name = assoc.name
if name not in implicit_output_dataset_collection_mapping:
implicit_output_dataset_collection_mapping[name] = []
implicit_output_dataset_collection_mapping[name].append(self.exported_key(assoc.dataset_collection))
job_attrs['input_dataset_mapping'] = input_dataset_mapping
job_attrs['input_dataset_collection_mapping'] = input_dataset_collection_mapping
job_attrs['input_dataset_collection_element_mapping'] = input_dataset_collection_element_mapping
job_attrs['output_dataset_mapping'] = output_dataset_mapping
job_attrs['output_dataset_collection_mapping'] = output_dataset_collection_mapping
job_attrs['implicit_output_dataset_collection_mapping'] = implicit_output_dataset_collection_mapping
jobs_attrs.append(job_attrs)
icjs_attrs = []
for icj in implicit_collection_jobs_dict.values():
icj_attrs = icj.serialize(self.security, self.serialization_options)
icjs_attrs.append(icj_attrs)
icjs_attrs_filename = os.path.join(export_directory, ATTRS_FILENAME_IMPLICIT_COLLECTION_JOBS)
with open(icjs_attrs_filename, 'w') as icjs_attrs_out:
icjs_attrs_out.write(json_encoder.encode(icjs_attrs))
export_attrs_filename = os.path.join(export_directory, ATTRS_FILENAME_EXPORT)
with open(export_attrs_filename, 'w') as export_attrs_out:
dump({"galaxy_export_version": GALAXY_EXPORT_VERSION}, export_attrs_out)
jobs_attrs_filename = os.path.join(export_directory, ATTRS_FILENAME_JOBS)
with open(jobs_attrs_filename, 'w') as jobs_attrs_out:
jobs_attrs_out.write(json_encoder.encode(jobs_attrs))
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is None:
self._finalize()