Merge pull request #21257 from nsoranzo/filter_job_search_by_history_id

Allow filtering job searches by history ID

Author: nsoranzo

client/src/api/schema/schema.ts

@@ -20056,6 +20056,11 @@ export interface components {
         };
         /** SearchJobsPayload */
         SearchJobsPayload: {
+            /**
+             * History ID
+             * @description The encoded ID of the history associated with this job.
+             */
+            history_id?: string | null;
             /**
              * Inputs
              * @description The inputs of the job.

lib/galaxy/managers/jobs.py

@@ -445,6 +445,7 @@ def by_tool_input(
         param: ToolStateJobInstancePopulatedT,
         param_dump: ToolStateDumpedToJsonInternalT,
         job_state: Optional[JobStatesT] = (Job.states.OK,),
+        history_id: Union[int, None] = None,
         require_name_match: bool = True,
     ):
         """Search for jobs producing same results using the 'inputs' part of a tool POST."""
@@ -489,6 +490,7 @@ def populate_input_data_input_id(path, key, value):
             job_state=job_state,
             param_dump=param_dump,
             wildcard_param_dump=wildcard_param_dump,
+            history_id=history_id,
             require_name_match=require_name_match,
         )
 
@@ -501,6 +503,7 @@ def __search(
         job_state: Optional[JobStatesT],
         param_dump: ToolStateDumpedToJsonInternalT,
         wildcard_param_dump=None,
+        history_id: Union[int, None] = None,
         require_name_match: bool = True,
     ):
         search_timer = ExecutionTimer()
@@ -566,7 +569,7 @@ def replace_dataset_ids(path, key, value):
                     return None
 
         stmt = stmt.where(*data_conditions).group_by(model.Job.id, *used_ids)
-        stmt = self._filter_jobs(stmt, tool_id, user.id, tool_version, job_state, wildcard_param_dump)
+        stmt = self._filter_jobs(stmt, tool_id, user.id, tool_version, job_state, wildcard_param_dump, history_id)
         stmt = self._exclude_jobs_with_deleted_outputs(stmt)
 
         for job in self.sa_session.execute(stmt):
@@ -634,8 +637,15 @@ def replace_dataset_ids(path, key, value):
         return None
 
     def _filter_jobs(
-        self, stmt, tool_id: str, user_id: int, tool_version: Optional[str], job_state, wildcard_param_dump
-    ):
+        self,
+        stmt: "Select[tuple[int]]",
+        tool_id: str,
+        user_id: int,
+        tool_version: Optional[str],
+        job_state: Union[JobStatesT, None],
+        wildcard_param_dump,
+        history_id: Union[int, None],
+    ) -> "Select[tuple[int]]":
         """Build subquery that selects a job with correct job parameters."""
         job_ids_materialized_cte = stmt.cte("job_ids_cte")
         outer_select_columns = [job_ids_materialized_cte.c[col.name] for col in stmt.selected_columns]
@@ -657,6 +667,9 @@ def _filter_jobs(
         if tool_version:
             stmt = stmt.where(Job.tool_version == str(tool_version))
 
+        if history_id is not None:
+            stmt = stmt.where(Job.history_id == history_id)
+
         if job_state is None:
             job_states: set[str] = {
                 Job.states.NEW,

lib/galaxy/schema/jobs.py

@@ -126,6 +126,11 @@ class SearchJobsPayload(Model):
         title="State",
         description="Current state of the job.",
     )
+    history_id: Union[DecodedDatabaseIdField, None] = Field(
+        default=None,
+        title="History ID",
+        description="The encoded ID of the history associated with this job.",
+    )
     model_config = ConfigDict(extra="allow")  # This is used for items named file_ and __file_
 
     @field_validator("inputs", mode="before")

lib/galaxy/webapps/galaxy/api/jobs.py

@@ -598,6 +598,7 @@ def search(
                 param=param,
                 param_dump=param_dump,
                 job_state=payload.state,
+                history_id=payload.history_id,
             )
             if job:
                 jobs.append(job)

lib/galaxy_test/api/test_jobs.py

@@ -4,6 +4,7 @@
 import time
 import urllib.parse
 from operator import itemgetter
+from typing import Union
 from unittest import SkipTest
 
 import pytest
@@ -903,9 +904,7 @@ def test_search_with_hdca_pair_input(self, history_id):
                 "f2": {"src": "hdca", "id": new_list_a},
             }
         )
-        search_payload = self._search_payload(
-            history_id=new_history_id, tool_id="multi_data_param", inputs=copied_inputs
-        )
+        search_payload = self._search_payload(history_id=history_id, tool_id="multi_data_param", inputs=copied_inputs)
         self._search(search_payload, expected_search_count=1)
         # Now we delete the original input HDCA that was used -- we should still be able to find the job
         delete_response = self._delete(f"histories/{history_id}/contents/dataset_collections/{list_id_a}")
@@ -1150,7 +1149,9 @@ def _job_search(self, tool_id, history_id, inputs):
         self._search(search_payload, expected_search_count=1)
         return tool_response
 
-    def _search_payload(self, history_id, tool_id, inputs, state="ok"):
+    def _search_payload(
+        self, tool_id: str, inputs: str, state: str = "ok", history_id: Union[str, None] = None
+    ) -> dict[str, Union[str, None]]:
         search_payload = dict(tool_id=tool_id, inputs=inputs, history_id=history_id, state=state)
         return search_payload