Merge pull request galaxyproject#21249 from mvdbeek/optimize_api_invocations_step_id

[25.1] Optimize /api/invocations/steps/{step_id}

Author: mvdbeek

lib/galaxy/model/__init__.py

@@ -1641,7 +1641,17 @@ class Job(Base, JobLike, UsesCreateAndUpdateTime, Dictifiable, Serializable):
         back_populates="job"
     )
 
-    dict_collection_visible_keys = ["id", "state", "exit_code", "update_time", "create_time", "galaxy_version"]
+    dict_collection_visible_keys = [
+        "id",
+        "state",
+        "exit_code",
+        "update_time",
+        "create_time",
+        "galaxy_version",
+        "tool_id",
+        "tool_version",
+        "history_id",
+    ]
     dict_element_visible_keys = [
         "id",
         "state",
@@ -1652,6 +1662,9 @@ class Job(Base, JobLike, UsesCreateAndUpdateTime, Dictifiable, Serializable):
         "command_version",
         "copied_from_job_id",
         "user_id",
+        "tool_id",
+        "tool_version",
+        "history_id",
     ]
 
     _numeric_metric = JobMetricNumeric
@@ -2188,9 +2201,6 @@ def to_dict(self, view="collection", system_details=False):
             rval = super().to_dict(view="collection")
         else:
             rval = super().to_dict(view=view)
-        rval["tool_id"] = self.tool_id
-        rval["tool_version"] = self.tool_version
-        rval["history_id"] = self.history_id
         if system_details or view == "admin_job_list":
             # System level details that only admins should have.
             rval["external_id"] = self.job_runner_external_id
@@ -2825,6 +2835,17 @@ def job_list(self):
             .all()
         )
 
+    def get_job_attributes(self, attributes: list[str]):
+        session = required_object_session(self)
+        targets = [getattr(Job.table.columns, attr) for attr in attributes]
+        stmt = (
+            select(*targets)
+            .select_from(Job)
+            .join(ImplicitCollectionJobsJobAssociation, Job.id == ImplicitCollectionJobsJobAssociation.job_id)
+            .where(ImplicitCollectionJobsJobAssociation.implicit_collection_jobs_id == self.id)
+        )
+        return session.execute(stmt)
+
     def _serialize(self, id_encoder, serialization_options):
         rval = dict_for(
             self,
@@ -10143,6 +10164,15 @@ def _serialize(self, id_encoder, serialization_options):
 
         return step_attrs
 
+    def get_jobs_dict(self):
+        if self.implicit_collection_jobs:
+            result = self.implicit_collection_jobs.get_job_attributes(Job.dict_collection_visible_keys)
+            return [{"model_class": "Job", **row._mapping} for row in result]
+        elif self.job:
+            return [self.job.to_dict()]
+        else:
+            return []
+
     def to_dict(self, view="collection", value_mapper=None):
         rval = super().to_dict(view=view, value_mapper=value_mapper)
         rval["order_index"] = self.workflow_step.order_index
@@ -10151,9 +10181,7 @@ def to_dict(self, view="collection", value_mapper=None):
         # Following no longer makes sense...
         # rval['state'] = self.job.state if self.job is not None else None
         if view == "element":
-            jobs = []
-            for job in self.jobs:
-                jobs.append(job.to_dict())
+            jobs = self.get_jobs_dict()
 
             outputs = {}
             for output_assoc in self.output_datasets:

lib/galaxy/tools/execute.py

@@ -541,7 +541,10 @@ def finalize_dataset_collections(self, trans):
                 # We mark this here so that the job cache query in subsequent
                 # jobs considers this to be a valid cached input.
                 completed_job_ids = {job.id for job in self.completed_jobs.values() if job}
-                if all(job.copied_from_job_id in completed_job_ids for job in self.implicit_collection_jobs.job_list):
+                completed_implicit_jobs_copied_from = set(
+                    self.implicit_collection_jobs.get_job_attributes(["copied_from_job_id"]).all()
+                )
+                if completed_implicit_jobs_copied_from <= completed_job_ids:
                     completed_collections = {
                         jtodca.name: jtodca.dataset_collection_instance
                         for jtodca in self.completed_jobs[0].output_dataset_collection_instances