Merge pull request #7471 from jmchilton/model_free_job_refactor_0

Collected Refactoring of Job Output Handling Code

lib/galaxy/jobs/__init__.py

@@ -1426,28 +1426,10 @@ class JobWrapper(HasResourceParameters):
         out_collections = dict([(obj.name, obj.dataset_collection_instance) for obj in job.output_dataset_collection_instances])
         out_collections.update([(obj.name, obj.dataset_collection) for obj in job.output_dataset_collections])
 
-        input_ext = 'data'
-        input_dbkey = '?'
-        for _, data in inp_data.items():
-            # For loop odd, but sort simulating behavior in galaxy.tools.actions
-            if not data:
-                continue
-            input_ext = data.ext
-            input_dbkey = data.dbkey or '?'
-        param_dict = self.get_param_dict(job)
-        # Create generated output children and primary datasets and dynamic outputs.
-        tool_working_directory = self.tool_working_directory
-        self.tool.collect_primary_datasets(out_data, self.get_tool_provided_job_metadata(), tool_working_directory, input_ext, input_dbkey)
-        self.tool.collect_dynamic_outputs(
-            out_collections,
-            self.get_tool_provided_job_metadata(),
-            job_working_directory=tool_working_directory,
-            inp_data=inp_data,
-            job=job,
-            input_dbkey=input_dbkey,
-        )
+        self.discover_outputs(job, inp_data, out_data, out_collections)
         # Certain tools require tasks to be completed after job execution
         # ( this used to be performed in the "exec_after_process" hook, but hooks are deprecated ).
+        param_dict = self.get_param_dict(job)
         self.tool.exec_after_process(self.app, inp_data, out_data, param_dict, job=job)
         # Call 'exec_after_process' hook
         self.tool.call_hook('exec_after_process', self.app, inp_data=inp_data,
@@ -1491,6 +1473,30 @@ class JobWrapper(HasResourceParameters):
         delete_files = cleanup_job == 'always' or (job.state == job.states.OK and cleanup_job == 'onsuccess')
         self.cleanup(delete_files=delete_files)
 
+    def discover_outputs(self, job, inp_data, out_data, out_collections):
+        input_ext = 'data'
+        input_dbkey = '?'
+        for _, data in inp_data.items():
+            # For loop odd, but sort simulating behavior in galaxy.tools.actions
+            if not data:
+                continue
+            input_ext = data.ext
+            input_dbkey = data.dbkey or '?'
+
+        # Create generated output children and primary datasets.
+        tool_working_directory = self.tool_working_directory
+        tool_provided_job_metadata = self.get_tool_provided_job_metadata()
+        self.tool.discover_outputs(
+            out_data,
+            out_collections,
+            tool_provided_job_metadata,
+            job=job,
+            tool_working_directory=tool_working_directory,
+            inp_data=inp_data,
+            input_ext=input_ext,
+            input_dbkey=input_dbkey,
+        )
+
     def check_tool_output(self, tool_stdout, tool_stderr, tool_exit_code, job, job_stdout=None, job_stderr=None):
         job_id_tag = "<unknown job id>"
         if job is not None:

lib/galaxy/tools/__init__.py

@@ -1681,17 +1681,32 @@ class Tool(Dictifiable):
         """
         pass
 
-    def collect_primary_datasets(self, output, tool_provided_metadata, job_working_directory, input_ext, input_dbkey="?"):
+    def discover_outputs(self, out_data, out_collections, tool_provided_metadata, tool_working_directory, job, input_ext, input_dbkey, inp_data=None):
         """
         Find any additional datasets generated by a tool and attach (for
         cases where number of outputs is not known in advance).
         """
-        return output_collect.collect_primary_datasets(self, output, tool_provided_metadata, job_working_directory, input_ext, input_dbkey=input_dbkey)
-
-    def collect_dynamic_outputs(self, output, tool_provided_metadata, **kwds):
-        """Collect dynamic outputs associated with a job from this tool.
-        """
-        return output_collect.collect_dynamic_outputs(self, output, tool_provided_metadata, **kwds)
+        tool = self
+        collected = output_collect.collect_primary_datasets(
+            tool,
+            out_data,
+            tool_provided_metadata,
+            tool_working_directory,
+            input_ext,
+            input_dbkey=input_dbkey,
+        )
+        output_collect.collect_dynamic_outputs(
+            tool,
+            out_collections,
+            tool_provided_metadata,
+            job_working_directory=tool_working_directory,
+            inp_data=inp_data,
+            job=job,
+            input_dbkey=input_dbkey,
+        )
+        # Return value only used in unit tests. Probably should be returning number of collected
+        # bytes instead?
+        return collected
 
     def to_archive(self):
         tool = self

lib/galaxy/tools/parameters/output_collect.py

@@ -6,6 +6,7 @@ import os
 import re
 from collections import namedtuple
 
+import galaxy.model
 from galaxy import util
 from galaxy.dataset_collections.structure import UninitializedTree
 from galaxy.tools.parser.output_collection_def import (
@@ -105,7 +106,7 @@ def collect_dynamic_outputs(
             object_id = destination.get("object_id")
             if object_id:
                 sa_session = tool.app.model.context
-                hdca = sa_session.query(app.model.HistoryDatasetCollectionAssociation).get(int(object_id))
+                hdca = sa_session.query(galaxy.model.HistoryDatasetCollectionAssociation).get(int(object_id))
             else:
                 name = unnamed_output_dict.get("name", "unnamed collection")
                 collection_type = unnamed_output_dict["collection_type"]
@@ -162,7 +163,7 @@ def collect_dynamic_outputs(
                         primary_dataset = None
                         if hda_id:
                             sa_session = tool.app.model.context
-                            primary_dataset = sa_session.query(app.model.HistoryDatasetAssociation).get(hda_id)
+                            primary_dataset = sa_session.query(galaxy.model.HistoryDatasetAssociation).get(hda_id)
 
                         dataset = job_context.create_dataset(
                             ext=ext,
@@ -306,7 +307,6 @@ class JobContext(object):
             )
             element_datasets.append((element_identifiers, dataset))
 
-        app = self.app
         sa_session = self.sa_session
         job = self.job
 
@@ -330,7 +330,7 @@ class JobContext(object):
             if job:
                 element_identifier_str = ":".join(element_identifiers)
                 # Below was changed from '__new_primary_file_%s|%s__' % (name, designation )
-                assoc = app.model.JobToOutputDatasetAssociation('__new_primary_file_%s|%s__' % (name, element_identifier_str), dataset)
+                assoc = galaxy.model.JobToOutputDatasetAssociation('__new_primary_file_%s|%s__' % (name, element_identifier_str), dataset)
                 assoc.job = self.job
             sa_session.add(assoc)
 
@@ -366,11 +366,6 @@ class JobContext(object):
             primary_data.visible = visible
             primary_data.dbkey = dbkey
 
-        # Copy metadata from one of the inputs if requested.
-        metadata_source = None
-        if metadata_source_name:
-            metadata_source = self.inp_data[metadata_source_name]
-
         sa_session.flush()
 
         if tag_list:
@@ -388,6 +383,11 @@ class JobContext(object):
         # designation.
         primary_data.name = name
 
+        # Copy metadata from one of the inputs if requested.
+        metadata_source = None
+        if metadata_source_name:
+            metadata_source = self.inp_data[metadata_source_name]
+
         if metadata_source:
             primary_data.init_meta(copy_from=metadata_source)
         else:
@@ -459,7 +459,7 @@ def collect_primary_datasets(tool, output, tool_provided_metadata, job_working_d
                 job = assoc.job
                 break
             if job:
-                assoc = app.model.JobToOutputDatasetAssociation('__new_primary_file_%s|%s__' % (name, designation), primary_data)
+                assoc = galaxy.model.JobToOutputDatasetAssociation('__new_primary_file_%s|%s__' % (name, designation), primary_data)
                 assoc.job = job
                 sa_session.add(assoc)
                 sa_session.flush()
@@ -758,18 +758,18 @@ def _new_ldda(
     dbkey,
     library_folder,
 ):
-    ld = trans.app.model.LibraryDataset(folder=library_folder, name=name)
+    ld = galaxy.model.LibraryDataset(folder=library_folder, name=name)
     trans.sa_session.add(ld)
     trans.sa_session.flush()
     trans.app.security_agent.copy_library_permissions(trans, library_folder, ld)
 
-    ldda = trans.app.model.LibraryDatasetDatasetAssociation(name=name,
-                                                            extension=ext,
-                                                            dbkey=dbkey,
-                                                            library_dataset=ld,
-                                                            user=trans.user,
-                                                            create_dataset=True,
-                                                            sa_session=trans.sa_session)
+    ldda = galaxy.model.LibraryDatasetDatasetAssociation(name=name,
+                                                         extension=ext,
+                                                         dbkey=dbkey,
+                                                         library_dataset=ld,
+                                                         user=trans.user,
+                                                         create_dataset=True,
+                                                         sa_session=trans.sa_session)
     trans.sa_session.add(ldda)
     ldda.state = ldda.states.OK
     # Permissions must be the same on the LibraryDatasetDatasetAssociation and the associated LibraryDataset
@@ -798,13 +798,13 @@ def _new_hda(
     """Return a new unflushed HDA with dataset and permissions setup.
     """
     # Create new primary dataset
-    primary_data = app.model.HistoryDatasetAssociation(extension=ext,
-                                                       designation=designation,
-                                                       visible=visible,
-                                                       dbkey=dbkey,
-                                                       create_dataset=True,
-                                                       flush=False,
-                                                       sa_session=sa_session)
+    primary_data = galaxy.model.HistoryDatasetAssociation(extension=ext,
+                                                          designation=designation,
+                                                          visible=visible,
+                                                          dbkey=dbkey,
+                                                          create_dataset=True,
+                                                          flush=False,
+                                                          sa_session=sa_session)
     if permissions is not UNSET:
         app.security_agent.set_all_dataset_permissions(primary_data.dataset, permissions, new=True, flush=False)
     sa_session.add(primary_data)

test/unit/tools/test_collect_primary_datasets.py

@@ -331,7 +331,7 @@ class CollectPrimaryDatasetsTestCase(unittest.TestCase, tools_support.UsesApp, t
         else:
             tool_provided_metadata = LegacyToolProvidedMetadata(meta_file)
 
-        return self.tool.collect_primary_datasets(self.outputs, tool_provided_metadata, job_working_directory, "txt", input_dbkey="btau")
+        return self.tool.discover_outputs(self.outputs, {}, tool_provided_metadata, job_working_directory, job=None, input_ext="txt", input_dbkey="btau")
 
     def _replace_output_collectors(self, xml_str):
         # Rewrite tool as if it had been created with output containing