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

@@ -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:

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)

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: