Fix SA2.0 (query->select) in webapps.galaxy.api

lib/galaxy/webapps/galaxy/api/forms.py

@@ -3,6 +3,8 @@
 """
 import logging
 
+from sqlalchemy import select
+
 from galaxy import web
 from galaxy.forms.forms import form_factory
 from galaxy.model.base import transaction
@@ -23,9 +25,10 @@ def index(self, trans, **kwd):
         if not trans.user_is_admin:
             trans.response.status = 403
             return "You are not authorized to view the list of forms."
-        query = trans.sa_session.query(trans.app.model.FormDefinition)
+
         rval = []
-        for form_definition in query:
+        form_defs = trans.sa_session.scalars(select(trans.app.model.FormDefinition)).all()
+        for form_definition in form_defs:
             item = form_definition.to_dict(
                 value_mapper={"id": trans.security.encode_id, "form_definition_current_id": trans.security.encode_id}
             )
@@ -46,7 +49,7 @@ def show(self, trans, id, **kwd):
             trans.response.status = 400
             return f"Malformed form definition id ( {str(form_definition_id)} ) specified, unable to decode."
         try:
-            form_definition = trans.sa_session.query(trans.app.model.FormDefinition).get(decoded_form_definition_id)
+            form_definition = trans.sa_session.get(trans.app.model.FormDefinition, decoded_form_definition_id)
         except Exception:
             form_definition = None
         if not form_definition or not trans.user_is_admin:

lib/galaxy/webapps/galaxy/api/job_files.py

@@ -126,7 +126,7 @@ def __authorize_job_access(self, trans, encoded_job_id, **kwargs):
             raise exceptions.ItemAccessibilityException("Invalid job_key supplied.")
 
         # Verify job is active. Don't update the contents of complete jobs.
-        job = trans.sa_session.query(model.Job).get(job_id)
+        job = trans.sa_session.get(model.Job, job_id)
         if job.finished:
             error_message = "Attempting to read or modify the files of a job that has already completed."
             raise exceptions.ItemAccessibilityException(error_message)

lib/galaxy/webapps/galaxy/api/job_tokens.py

@@ -53,15 +53,15 @@ def get_token(
 
     def __authorize_job_access(self, trans, encoded_job_id, job_key):
         job_id = trans.security.decode_id(encoded_job_id)
-        job = trans.sa_session.query(model.Job).get(job_id)
+        job = trans.sa_session.get(model.Job, job_id)
         secret = job.destination_params.get("job_secret_base", "jobs_token")
 
         job_key_internal = trans.security.encode_id(job_id, kind=secret)
         if not util.safe_str_cmp(job_key_internal, job_key):
             raise exceptions.AuthenticationFailed("Invalid job_key supplied.")
 
         # Verify job is active
-        job = trans.sa_session.query(model.Job).get(job_id)
+        job = trans.sa_session.get(model.Job, job_id)
         if job.finished:
             error_message = "Attempting to get oidc token for a job that has already completed."
             raise exceptions.ItemAccessibilityException(error_message)

lib/galaxy/webapps/galaxy/api/library_contents.py

@@ -4,6 +4,7 @@
 import logging
 from typing import Optional
 
+from sqlalchemy import select
 from sqlalchemy.orm.exc import (
     MultipleResultsFound,
     NoResultFound,
@@ -103,11 +104,8 @@ def traverse(folder):
 
         decoded_library_id = self.decode_id(library_id)
         try:
-            library = (
-                trans.sa_session.query(trans.app.model.Library)
-                .filter(trans.app.model.Library.table.c.id == decoded_library_id)
-                .one()
-            )
+            stmt = select(trans.app.model.Library).filter(trans.app.model.Library.id == decoded_library_id)
+            library = trans.sa_session.execute(stmt).scalar_one()
         except MultipleResultsFound:
             raise exceptions.InconsistentDatabase("Multiple libraries found with the same id.")
         except NoResultFound:
@@ -348,7 +346,7 @@ def _upload_library_dataset(self, trans, folder_id: int, **kwd):
         roles = kwd.get("roles", "")
         is_admin = trans.user_is_admin
         current_user_roles = trans.get_current_user_roles()
-        folder = trans.sa_session.query(trans.app.model.LibraryFolder).get(folder_id)
+        folder = trans.sa_session.get(trans.app.model.LibraryFolder, folder_id)
         self._check_access(trans, is_admin, folder, current_user_roles)
         self._check_add(trans, is_admin, folder, current_user_roles)
         library = folder.parent_library

lib/galaxy/webapps/galaxy/api/page_revisions.py

@@ -3,6 +3,8 @@
 """
 import logging
 
+from sqlalchemy import select
+
 from galaxy.managers.base import get_object
 from galaxy.managers.pages import PageManager
 from galaxy.web import expose_api
@@ -30,7 +32,7 @@ def index(self, trans, page_id, **kwd):
         :returns:   dictionaries containing different revisions of the page
         """
         page = get_object(trans, page_id, "Page", check_ownership=False, check_accessible=True)
-        r = trans.sa_session.query(trans.app.model.PageRevision).filter_by(page_id=page.id)
+        r = trans.sa_session.scalars(select(trans.app.model.PageRevision).filter_by(page_id=page.id))
         out = []
         for page in r:
             as_dict = self.encode_all_ids(trans, page.to_dict(), True)

lib/galaxy/webapps/galaxy/api/tool_entry_points.py

@@ -51,7 +51,7 @@ def index(self, trans: ProvidesUserContext, running=False, job_id=None, **kwd):
             )
 
         if job_id is not None:
-            job = trans.sa_session.query(Job).get(self.decode_id(job_id))
+            job = trans.sa_session.get(Job, self.decode_id(job_id))
             if not self.interactivetool_manager.can_access_job(trans, job):
                 raise exceptions.ItemAccessibilityException()
             entry_points = job.interactivetool_entry_points
@@ -94,7 +94,7 @@ def stop_entry_point(self, trans: ProvidesUserContext, id, **kwds):
             raise exceptions.RequestParameterMissingException("Must supply entry point id")
         try:
             entry_point_id = self.decode_id(id)
-            entry_point = trans.sa_session.query(InteractiveToolEntryPoint).get(entry_point_id)
+            entry_point = trans.sa_session.get(InteractiveToolEntryPoint, entry_point_id)
         except Exception:
             raise exceptions.RequestParameterInvalidException("entry point invalid")
         if self.app.interactivetool_manager.can_access_entry_point(trans, entry_point):

lib/galaxy/webapps/galaxy/api/users.py

@@ -21,6 +21,7 @@
 from sqlalchemy import (
     false,
     or_,
+    select,
     true,
 )
 from typing_extensions import Literal
@@ -292,31 +293,31 @@ def index(self, trans: ProvidesUserContext, deleted="False", f_email=None, f_nam
         :type  f_any: str
         """
         rval = []
-        query = trans.sa_session.query(User)
+        stmt = select(User)
         deleted = util.string_as_bool(deleted)
 
         if f_email and (trans.user_is_admin or trans.app.config.expose_user_email):
-            query = query.filter(User.email.like(f"%{f_email}%"))
+            stmt = stmt.filter(User.email.like(f"%{f_email}%"))
 
         if f_name and (trans.user_is_admin or trans.app.config.expose_user_name):
-            query = query.filter(User.username.like(f"%{f_name}%"))
+            stmt = stmt.filter(User.username.like(f"%{f_name}%"))
 
         if f_any:
             if trans.user_is_admin:
-                query = query.filter(or_(User.email.like(f"%{f_any}%"), User.username.like(f"%{f_any}%")))
+                stmt = stmt.filter(or_(User.email.like(f"%{f_any}%"), User.username.like(f"%{f_any}%")))
             else:
                 if trans.app.config.expose_user_email and trans.app.config.expose_user_name:
-                    query = query.filter(or_(User.email.like(f"%{f_any}%"), User.username.like(f"%{f_any}%")))
+                    stmt = stmt.filter(or_(User.email.like(f"%{f_any}%"), User.username.like(f"%{f_any}%")))
                 elif trans.app.config.expose_user_email:
-                    query = query.filter(User.email.like(f"%{f_any}%"))
+                    stmt = stmt.filter(User.email.like(f"%{f_any}%"))
                 elif trans.app.config.expose_user_name:
-                    query = query.filter(User.username.like(f"%{f_any}%"))
+                    stmt = stmt.filter(User.username.like(f"%{f_any}%"))
 
         if deleted:
             # only admins can see deleted users
             if not trans.user_is_admin:
                 return []
-            query = query.filter(User.table.c.deleted == true())
+            stmt = stmt.filter(User.deleted == true())
         else:
             # special case: user can see only their own user
             # special case2: if the galaxy admin has specified that other user email/names are
@@ -328,8 +329,8 @@ def index(self, trans: ProvidesUserContext, deleted="False", f_email=None, f_nam
             ):
                 item = trans.user.to_dict(value_mapper={"id": trans.security.encode_id})
                 return [item]
-            query = query.filter(User.table.c.deleted == false())
-        for user in query:
+            stmt = stmt.filter(User.deleted == false())
+        for user in trans.sa_session.scalars(stmt).all():
             item = user.to_dict(value_mapper={"id": trans.security.encode_id})
             # If NOT configured to expose_email, do not expose email UNLESS the user is self, or
             # the user is an admin
@@ -699,8 +700,8 @@ def set_information(self, trans, id, payload=None, **kwd):
         user_info_form_id = payload.get("info|form_id")
         if user_info_form_id:
             prefix = "info|"
-            user_info_form = trans.sa_session.query(trans.app.model.FormDefinition).get(
-                trans.security.decode_id(user_info_form_id)
+            user_info_form = trans.sa_session.get(
+                trans.app.model.FormDefinition, trans.security.decode_id(user_info_form_id)
             )
             user_info_values = {}
             for item in payload:
@@ -752,7 +753,7 @@ def set_information(self, trans, id, payload=None, **kwd):
             d = address_dicts[index]
             if d.get("id"):
                 try:
-                    user_address = trans.sa_session.query(UserAddress).get(trans.security.decode_id(d["id"]))
+                    user_address = trans.sa_session.get(UserAddress, trans.security.decode_id(d["id"]))
                 except Exception as e:
                     raise exceptions.ObjectNotFound(f"Failed to access user address ({d['id']}). {e}")
             else:
@@ -921,9 +922,7 @@ def set_permissions(self, trans, id, payload=None, **kwd):
         permissions = {}
         for index, action in trans.app.model.Dataset.permitted_actions.items():
             action_id = trans.app.security_agent.get_action(action.action).action
-            permissions[action_id] = [
-                trans.sa_session.query(trans.app.model.Role).get(x) for x in (payload.get(index) or [])
-            ]
+            permissions[action_id] = [trans.sa_session.get(trans.app.model.Role, x) for x in (payload.get(index) or [])]
         trans.app.security_agent.user_set_default_permissions(user, permissions)
         return {"message": "Permissions have been saved."}
 
@@ -1042,8 +1041,8 @@ def get_custom_builds(self, trans, id, payload=None, **kwd):
         update = False
         for key, dbkey in dbkeys.items():
             if "count" not in dbkey and "linecount" in dbkey:
-                chrom_count_dataset = trans.sa_session.query(trans.app.model.HistoryDatasetAssociation).get(
-                    dbkey["linecount"]
+                chrom_count_dataset = trans.sa_session.get(
+                    trans.app.model.HistoryDatasetAssociation, dbkey["linecount"]
                 )
                 if (
                     chrom_count_dataset
@@ -1139,7 +1138,7 @@ def add_custom_builds(self, trans, id, key, payload=None, **kwd):
                 build_dict["count"] = counter
             else:
                 build_dict["fasta"] = trans.security.decode_id(len_value)
-                dataset = trans.sa_session.query(trans.app.model.HistoryDatasetAssociation).get(build_dict["fasta"])
+                dataset = trans.sa_session.get(trans.app.model.HistoryDatasetAssociation, build_dict["fasta"])
                 try:
                     new_len = dataset.get_converted_dataset(trans, "len")
                     new_linecount = new_len.get_converted_dataset(trans, "linecount")

lib/galaxy/webapps/galaxy/api/workflows.py

@@ -23,6 +23,10 @@
 from gxformat2._yaml import ordered_dump
 from markupsafe import escape
 from pydantic import Extra
+from sqlalchemy import (
+    func,
+    select,
+)
 from starlette.responses import StreamingResponse
 
 from galaxy import (
@@ -162,7 +166,6 @@ def set_workflow_menu(self, trans: GalaxyWebTransaction, payload=None, **kwd):
         for m in user.stored_workflow_menu_entries:
             sess.delete(m)
         user.stored_workflow_menu_entries = []
-        q = sess.query(model.StoredWorkflow)
         # To ensure id list is unique
         seen_workflow_ids = set()
         for wf_id in workflow_ids_decoded:
@@ -171,7 +174,8 @@ def set_workflow_menu(self, trans: GalaxyWebTransaction, payload=None, **kwd):
             else:
                 seen_workflow_ids.add(wf_id)
             m = model.StoredWorkflowMenuEntry()
-            m.stored_workflow = q.get(wf_id)
+            m.stored_workflow = sess.get(model.StoredWorkflow, wf_id)
+
             user.stored_workflow_menu_entries.append(m)
         with transaction(sess):
             sess.commit()
@@ -192,12 +196,11 @@ def show(self, trans: GalaxyWebTransaction, id, **kwd):
         """
         stored_workflow = self.__get_stored_workflow(trans, id, **kwd)
         if stored_workflow.importable is False and stored_workflow.user != trans.user and not trans.user_is_admin:
-            if (
-                trans.sa_session.query(model.StoredWorkflowUserShareAssociation)
-                .filter_by(user=trans.user, stored_workflow=stored_workflow)
-                .count()
-                == 0
-            ):
+            stmt = select(model.StoredWorkflowUserShareAssociation).filter_by(
+                user=trans.user, stored_workflow=stored_workflow
+            )
+            stmt = select(func.count()).select_from(stmt)
+            if trans.sa_session.scalar(stmt) == 0:
                 message = "Workflow is neither importable, nor owned by or shared with current user"
                 raise exceptions.ItemAccessibilityException(message)
         if kwd.get("legacy", False):