Convert Query to Select

lib/galaxy/managers/jobs.py

@@ -120,59 +120,53 @@ class JobManager:
             if decoded_user_id is not None and decoded_user_id != trans.user.id:
                 raise AdminRequiredException("Only admins can index the jobs of others")
 
+        stmt = select(Job)
+
         if is_admin:
             if decoded_user_id is not None:
-                query = trans.sa_session.query(Job).filter(Job.user_id == decoded_user_id)
-            else:
-                query = trans.sa_session.query(Job)
+                stmt = stmt.where(Job.user_id == decoded_user_id)
             if user_details:
-                query = query.outerjoin(Job.user)
+                stmt = stmt.outerjoin(Job.user)
         else:
-            query = trans.sa_session.query(Job).filter(Job.user_id == trans.user.id)
+            stmt = stmt.where(Job.user_id == trans.user.id)
 
-        def build_and_apply_filters(query, objects, filter_func):
+        def build_and_apply_filters(stmt, objects, filter_func):
             if objects is not None:
                 if isinstance(objects, (str, date, datetime)):
-                    query = query.filter(filter_func(objects))
+                    stmt = stmt.where(filter_func(objects))
                 elif isinstance(objects, list):
                     t = []
                     for obj in objects:
                         t.append(filter_func(obj))
-                    query = query.filter(or_(*t))
-            return query
+                    stmt = stmt.where(or_(*t))
+            return stmt
 
-        query = build_and_apply_filters(query, payload.states, lambda s: Job.state == s)
-        query = build_and_apply_filters(query, payload.tool_ids, lambda t: Job.tool_id == t)
-        query = build_and_apply_filters(query, payload.tool_ids_like, lambda t: Job.tool_id.like(t))
-        query = build_and_apply_filters(query, payload.date_range_min, lambda dmin: Job.update_time >= dmin)
-        query = build_and_apply_filters(query, payload.date_range_max, lambda dmax: Job.update_time <= dmax)
+        stmt = build_and_apply_filters(stmt, payload.states, lambda s: model.Job.state == s)
+        stmt = build_and_apply_filters(stmt, payload.tool_ids, lambda t: model.Job.tool_id == t)
+        stmt = build_and_apply_filters(stmt, payload.tool_ids_like, lambda t: model.Job.tool_id.like(t))
+        stmt = build_and_apply_filters(stmt, payload.date_range_min, lambda dmin: model.Job.update_time >= dmin)
+        stmt = build_and_apply_filters(stmt, payload.date_range_max, lambda dmax: model.Job.update_time <= dmax)
 
         if history_id is not None:
-            query = query.filter(Job.history_id == history_id)
+            stmt = stmt.where(Job.history_id == history_id)
+
         if workflow_id or invocation_id:
+            wfi_step = select(WorkflowInvocationStep)
             if workflow_id is not None:
                 wfi_step = (
-                    trans.sa_session.query(WorkflowInvocationStep)
-                    .join(WorkflowInvocation)
-                    .join(Workflow)
-                    .filter(
-                        Workflow.stored_workflow_id == workflow_id,
-                    )
-                    .subquery()
+                    wfi_step.join(WorkflowInvocation).join(Workflow).where(Workflow.stored_workflow_id == workflow_id)
                 )
             elif invocation_id is not None:
-                wfi_step = (
-                    trans.sa_session.query(WorkflowInvocationStep)
-                    .filter(WorkflowInvocationStep.workflow_invocation_id == invocation_id)
-                    .subquery()
-                )
-            query1 = query.join(wfi_step)
-            query2 = query.join(ImplicitCollectionJobsJobAssociation).join(
+                wfi_step = wfi_step.where(WorkflowInvocationStep.workflow_invocation_id == invocation_id)
+            wfi_step = wfi_step.subquery()
+
+            stmt1 = stmt.join(wfi_step)
+            stmt2 = stmt.join(ImplicitCollectionJobsJobAssociation).join(
                 wfi_step,
                 ImplicitCollectionJobsJobAssociation.implicit_collection_jobs_id
                 == wfi_step.c.implicit_collection_jobs_id,
             )
-            query = query1.union(query2)
+            stmt = stmt1.union(stmt2)
 
         if search:
             search_filters = {
@@ -201,13 +195,13 @@ class JobManager:
                 if isinstance(term, FilteredTerm):
                     key = term.filter
                     if key == "user":
-                        query = query.filter(text_column_filter(User.email, term))
+                        stmt = stmt.where(text_column_filter(User.email, term))
                     elif key == "tool":
-                        query = query.filter(text_column_filter(Job.tool_id, term))
+                        stmt = stmt.where(text_column_filter(Job.tool_id, term))
                     elif key == "handler":
-                        query = query.filter(text_column_filter(Job.handler, term))
+                        stmt = stmt.where(text_column_filter(Job.handler, term))
                     elif key == "runner":
-                        query = query.filter(text_column_filter(Job.job_runner_name, term))
+                        stmt = stmt.where(text_column_filter(Job.job_runner_name, term))
                 elif isinstance(term, RawTextTerm):
                     columns = [Job.tool_id]
                     if user_details:
@@ -215,17 +209,17 @@ class JobManager:
                     if is_admin:
                         columns.append(Job.handler)
                         columns.append(Job.job_runner_name)
-                    query = query.filter(raw_text_column_filter(columns, term))
+                    stmt = stmt.filter(raw_text_column_filter(columns, term))
 
         if order_by == JobIndexSortByEnum.create_time:
             _order_by = Job.create_time.desc()
         else:
             _order_by = Job.update_time.desc()
-        query = query.order_by(_order_by)
+        stmt = stmt.order_by(_order_by)
 
-        query = query.offset(payload.offset)
-        query = query.limit(payload.limit)
-        return query
+        stmt = stmt.offset(payload.offset)
+        stmt = stmt.limit(payload.limit)
+        return trans.sa_session.scalars(stmt)
 
     def job_lock(self) -> JobLock:
         return JobLock(active=self.app.job_manager.job_lock)