From ee6ec264c4d606fe33818e2ec351fa17d309dc37 Mon Sep 17 00:00:00 2001 From: mvdbeek Date: Sat, 16 Oct 2021 17:11:17 +0200 Subject: [PATCH] Flush just once per import That might mean we accumulate a lot of items in memory, but if we don't do this we need to assign hids for every flush (which is possible). --- lib/galaxy/jobs/actions/post.py | 20 ++++++++++---------- lib/galaxy/model/store/__init__.py | 11 +---------- test/unit/test_model_store.py | 4 ++-- 3 files changed, 13 insertions(+), 22 deletions(-) diff --git a/lib/galaxy/jobs/actions/post.py b/lib/galaxy/jobs/actions/post.py index be0bd5a70cc..9444240bbd5 100644 --- a/lib/galaxy/jobs/actions/post.py +++ b/lib/galaxy/jobs/actions/post.py @@ -209,7 +209,7 @@ class RenameDatasetAction(DefaultJobAction): @classmethod def execute(cls, app, sa_session, action, job, replacement_dict, final_job_state=None): input_names = {} - # Lookp through inputs find one with "to_be_replaced" input + # Loop through inputs find one with "to_be_replaced" input # variable name, and get the replacement name for input_assoc in job.input_datasets: if input_assoc.dataset: @@ -399,31 +399,31 @@ class TagDatasetAction(DefaultJobAction): @classmethod def execute_on_mapped_over(cls, trans, sa_session, action, step_inputs, step_outputs, replacement_dict, final_job_state=None): + tag_handler = trans.app.tag_handler.create_tag_handler_session() if action.action_arguments: tags = [t.replace('#', 'name:') if t.startswith('#') else t for t in [t.strip() for t in action.action_arguments.get('tags', '').split(',') if t.strip()]] if tags: for name, step_output in step_outputs.items(): if action.output_name == '' or name == action.output_name: - cls._execute(trans.app, trans.user, step_output, tags) + cls._execute(tag_handler, trans.user, step_output, tags) @classmethod def execute(cls, app, sa_session, action, job, replacement_dict, final_job_state=None): if action.action_arguments: + tag_handler = app.tag_handler.create_tag_handler_session() tags = [t.replace('#', 'name:') if t.startswith('#') else t for t in [t.strip() for t in action.action_arguments.get('tags', '').split(',') if t.strip()]] if tags: for dataset_assoc in job.output_datasets: if action.output_name == '' or dataset_assoc.name == action.output_name: - cls._execute(app, job.user, dataset_assoc.dataset, tags) + cls._execute(tag_handler, job.user, dataset_assoc.dataset, tags) for dataset_collection_assoc in job.output_dataset_collection_instances: if action.output_name == '' or dataset_collection_assoc.name == action.output_name: - cls._execute(app, job.user, dataset_collection_assoc.dataset_collection_instance, tags) - - sa_session.flush() + cls._execute(tag_handler, job.user, dataset_collection_assoc.dataset_collection_instance, tags) @classmethod - def _execute(cls, app, user, output, tags): - app.tag_handler.add_tags_from_list(user, output, tags) + def _execute(cls, tag_handler, user, output, tags): + tag_handler.add_tags_from_list(user, output, tags) @classmethod def get_short_str(cls, pja): @@ -443,8 +443,8 @@ class RemoveTagDatasetAction(TagDatasetAction): direction = "from" @classmethod - def _execute(cls, app, user, output, tags): - app.tag_handler.remove_tags_from_list(user, output, tags) + def _execute(cls, tag_handler, user, output, tags): + tag_handler.remove_tags_from_list(user, output, tags) class ActionBox: diff --git a/lib/galaxy/model/store/__init__.py b/lib/galaxy/model/store/__init__.py index 7d99dc66048..3ff3ddbf3de 100644 --- a/lib/galaxy/model/store/__init__.py +++ b/lib/galaxy/model/store/__init__.py @@ -247,7 +247,6 @@ class ModelImportStore(metaclass=abc.ABCMeta): setattr(hda, attribute, value) handle_dataset_object_edit(hda) - self._flush() else: metadata = dataset_attrs['metadata'] @@ -295,7 +294,6 @@ class ModelImportStore(metaclass=abc.ABCMeta): dataset_instance.dataset.uuid = dataset_attrs["dataset_uuid"] self._session_add(dataset_instance) - self._flush() if model_class == "HistoryDatasetAssociation": # don't use add_history to manage HID handling across full import to try to preserve @@ -306,7 +304,6 @@ class ModelImportStore(metaclass=abc.ABCMeta): else: object_import_tracker.requires_hid.append(dataset_instance) - self._flush() if 'dataset' in dataset_attrs: handle_dataset_object_edit(dataset_instance) else: @@ -404,7 +401,6 @@ class ModelImportStore(metaclass=abc.ABCMeta): library_folder.deleted = deleted self._session_add(library_folder) - self._flush() for sub_folder_attrs in folder_attrs.get("folders", []): sub_folder = import_folder(sub_folder_attrs) @@ -422,8 +418,7 @@ class ModelImportStore(metaclass=abc.ABCMeta): ld.library_dataset_dataset_association = ldda self._session_add(ld) - self._flush() - + self.sa_session.flush() return library_folder if 'root_folder' in library_attrs: @@ -657,7 +652,6 @@ class ModelImportStore(metaclass=abc.ABCMeta): assert not self.sessionless job = self.sa_session.query(model.Job).get(job_attrs["id"]) self._connect_job_io(job, job_attrs, _find_hda, _find_hdca, _find_dce) - self._flush() continue imported_job = model.Job() @@ -691,7 +685,6 @@ class ModelImportStore(metaclass=abc.ABCMeta): except Exception: pass self._session_add(imported_job) - self._flush() # Connect jobs to input and output datasets. params = self._normalize_job_parameters(imported_job, job_attrs, _find_hda, _find_hdca, _find_dce) @@ -700,7 +693,6 @@ class ModelImportStore(metaclass=abc.ABCMeta): imported_job.add_parameter(name, dumps(value)) self._connect_job_io(imported_job, job_attrs, _find_hda, _find_hdca, _find_dce) - self._flush() if object_key in job_attrs: object_import_tracker.jobs_by_key[job_attrs[object_key]] = imported_job @@ -722,7 +714,6 @@ class ModelImportStore(metaclass=abc.ABCMeta): self._session_add(icja) self._session_add(icj) - self._flush() def _session_add(self, obj): self.sa_session.add(obj) diff --git a/test/unit/test_model_store.py b/test/unit/test_model_store.py index d180fe71176..e891dd0cd93 100644 --- a/test/unit/test_model_store.py +++ b/test/unit/test_model_store.py @@ -221,10 +221,9 @@ def test_import_export_edit_collection(): sa_session.add(hc1) sa_session.add(h) - sa_session.flush() - import_history = model.History(name="Test History for Import", user=u) sa_session.add(import_history) + sa_session.flush() temp_directory = mkdtemp() with store.DirectoryModelExportStore(temp_directory, app=app, for_edit=True) as export_store: @@ -404,6 +403,7 @@ def _setup_simple_export(export_kwds): import_history = model.History(name="Test History for Import", user=u) sa_session.add(import_history) + sa_session.flush() temp_directory = mkdtemp() with store.DirectoryModelExportStore(temp_directory, app=app, **export_kwds) as export_store: