Fix SA2.0 ORM usage in webapps.galaxy.services

Move data access method to managers.quotas (get_quotas)
Move data access method to managers.histories (get_len_files_by_history)
Move data access method to managers.groups (get_group_by_name)

lib/galaxy/managers/histories.py

@@ -24,6 +24,7 @@
     select,
     true,
 )
+from sqlalchemy.orm import Session
 from typing_extensions import Literal
 
 from galaxy import (
@@ -43,6 +44,7 @@
     StorageCleanerManager,
 )
 from galaxy.managers.export_tracker import StoreExportTracker
+from galaxy.model import HistoryDatasetAssociation
 from galaxy.model.base import transaction
 from galaxy.schema.fields import DecodedDatabaseIdField
 from galaxy.schema.schema import (
@@ -900,3 +902,12 @@ def username_eq(self, item, val: str) -> bool:
 
     def username_contains(self, item, val: str) -> bool:
         return val.lower() in str(item.user.username).lower()
+
+
+def get_fasta_hdas_by_history(session: Session, history_id: int):
+    stmt = (
+        select(HistoryDatasetAssociation)
+        .filter_by(history_id=history_id, extension="fasta", deleted=False)
+        .order_by(HistoryDatasetAssociation.hid.desc())
+    )
+    return session.scalars(stmt).all()

lib/galaxy/managers/quotas.py

@@ -11,12 +11,20 @@
     Union,
 )
 
+from sqlalchemy import (
+    false,
+    select,
+    true,
+)
+from sqlalchemy.orm import Session
+
 from galaxy import (
     model,
     util,
 )
 from galaxy.exceptions import ActionInputError
 from galaxy.managers import base
+from galaxy.model import Quota
 from galaxy.model.base import transaction
 from galaxy.quota import DatabaseQuotaAgent
 from galaxy.quota._schema import (
@@ -273,3 +281,11 @@ def purge_quota(self, quota, params=None):
 
     def get_quota(self, trans, id: int, deleted: Optional[bool] = None) -> model.Quota:
         return base.get_object(trans, id, "Quota", check_ownership=False, check_accessible=False, deleted=deleted)
+
+
+def get_quotas(session: Session, deleted: bool = False):
+    is_deleted = true()
+    if not deleted:
+        is_deleted = false()
+    stmt = select(Quota).where(Quota.deleted == is_deleted)
+    return session.scalars(stmt)

lib/galaxy/webapps/galaxy/services/dataset_collections.py

@@ -210,7 +210,7 @@ def show(
         return rval
 
     def dce_content(self, trans: ProvidesHistoryContext, dce_id: DecodedDatabaseIdField) -> DCESummary:
-        dce: Optional[DatasetCollectionElement] = trans.model.session.query(DatasetCollectionElement).get(dce_id)
+        dce: Optional[DatasetCollectionElement] = trans.model.session.get(DatasetCollectionElement, dce_id)
         if not dce:
             raise exceptions.ObjectNotFound("No DatasetCollectionElement found")
         if not trans.user_is_admin:

lib/galaxy/webapps/galaxy/services/histories.py

@@ -32,6 +32,7 @@
 from galaxy.managers.citations import CitationsManager
 from galaxy.managers.context import ProvidesHistoryContext
 from galaxy.managers.histories import (
+    get_fasta_hdas_by_history,
     HistoryDeserializer,
     HistoryExportManager,
     HistoryFilters,
@@ -640,11 +641,7 @@ def get_custom_builds_metadata(
         installed_builds = []
         for build in glob.glob(os.path.join(trans.app.config.len_file_path, "*.len")):
             installed_builds.append(os.path.basename(build).split(".len")[0])
-        fasta_hdas = (
-            trans.sa_session.query(model.HistoryDatasetAssociation)
-            .filter_by(history=history, extension="fasta", deleted=False)
-            .order_by(model.HistoryDatasetAssociation.hid.desc())
-        )
+        fasta_hdas = get_fasta_hdas_by_history(trans.sa_session, history.id)
         return CustomBuildsMetadataResponse(
             installed_builds=[LabelValuePair(label=ins, value=ins) for ins in installed_builds],
             fasta_hdas=[

lib/galaxy/webapps/galaxy/services/libraries.py

@@ -15,6 +15,7 @@
 from galaxy.managers.folders import FolderManager
 from galaxy.managers.libraries import LibraryManager
 from galaxy.managers.roles import RoleManager
+from galaxy.model import Role
 from galaxy.schema.fields import DecodedDatabaseIdField
 from galaxy.schema.schema import (
     BasicRoleModel,
@@ -321,7 +322,7 @@ def set_permissions_old(self, trans, library, payload: Dict[str, Any]) -> Librar
         permissions = {}
         for k, v in trans.app.model.Library.permitted_actions.items():
             role_params = payload.get(f"{k}_in", [])
-            in_roles = [trans.sa_session.query(trans.app.model.Role).get(x) for x in util.listify(role_params)]
+            in_roles = [trans.sa_session.get(Role, x) for x in util.listify(role_params)]
             permissions[trans.app.security_agent.get_action(v.action)] = in_roles
         trans.app.security_agent.set_all_library_permissions(trans, library, permissions)
         trans.sa_session.refresh(library)

lib/galaxy/webapps/galaxy/services/quotas.py

@@ -1,17 +1,14 @@
 import logging
 from typing import Optional
 
-from sqlalchemy import (
-    false,
-    true,
-)
-
-from galaxy import (
-    model,
-    util,
-)
+from galaxy import util
 from galaxy.managers.context import ProvidesUserContext
-from galaxy.managers.quotas import QuotaManager
+from galaxy.managers.groups import get_group_by_name
+from galaxy.managers.quotas import (
+    get_quotas,
+    QuotaManager,
+)
+from galaxy.model.repositories import get_user_by_email
 from galaxy.quota._schema import (
     CreateQuotaParams,
     CreateQuotaResult,
@@ -39,14 +36,13 @@ def __init__(self, security: IdEncodingHelper, quota_manager: QuotaManager):
     def index(self, trans: ProvidesUserContext, deleted: bool = False) -> QuotaSummaryList:
         """Displays a list of quotas."""
         rval = []
-        query = trans.sa_session.query(model.Quota)
         if deleted:
             route = "deleted_quota"
-            query = query.filter(model.Quota.deleted == true())
+            quotas = get_quotas(trans.sa_session, deleted=True)
         else:
             route = "quota"
-            query = query.filter(model.Quota.deleted == false())
-        for quota in query:
+            quotas = get_quotas(trans.sa_session, deleted=False)
+        for quota in quotas:
             item = quota.to_dict(value_mapper={"id": DecodedDatabaseIdField.encode})
             encoded_id = DecodedDatabaseIdField.encode(quota.id)
             item["url"] = url_for(route, id=encoded_id)
@@ -119,25 +115,29 @@ def validate_in_users_and_groups(self, trans, payload):
         For convenience, in_users and in_groups can be encoded IDs or emails/group names in the API.
         """
 
-        def get_id(item, model_class, column):
+        def get_user_id(item):
+            try:
+                return trans.security.decode_id(item)
+            except Exception:
+                return user_repo.get_by_email(item).id
+
+        def get_group_id(item):
             try:
                 return trans.security.decode_id(item)
             except Exception:
-                pass  # maybe an email/group name
-            # this will raise if the item is invalid
-            return trans.sa_session.query(model_class).filter(column == item).first().id
+                return get_group_by_name(trans.sa_session, item).id
 
         new_in_users = []
         new_in_groups = []
         invalid = []
         for item in util.listify(payload.get("in_users", [])):
             try:
-                new_in_users.append(get_id(item, model.User, model.User.email))
+                new_in_users.append(get_user_id(item))
             except Exception:
                 invalid.append(item)
         for item in util.listify(payload.get("in_groups", [])):
             try:
-                new_in_groups.append(get_id(item, model.Group, model.Group.name))
+                new_in_groups.append(get_group_id(item))
             except Exception:
                 invalid.append(item)
         if invalid:

lib/galaxy/webapps/galaxy/services/tool_shed_repositories.py

@@ -5,9 +5,9 @@
 
 from pydantic import BaseModel
 from sqlalchemy import (
-    and_,
     cast,
     Integer,
+    select,
 )
 
 from galaxy.model.scoped_session import install_model_scoped_session
@@ -17,10 +17,7 @@
     CheckForUpdatesResponse,
     InstalledToolShedRepository,
 )
-from galaxy.tool_shed.util.repository_util import (
-    check_for_updates,
-    get_tool_shed_repository_by_decoded_id,
-)
+from galaxy.tool_shed.util.repository_util import check_for_updates
 from galaxy.util.tool_shed.tool_shed_registry import Registry
 from galaxy.web import url_for
 
@@ -43,31 +40,20 @@ def __init__(
         self._tool_shed_registry = tool_shed_registry
 
     def index(self, request: InstalledToolShedRepositoryIndexRequest) -> List[InstalledToolShedRepository]:
-        clause_list = []
-        if request.name is not None:
-            clause_list.append(ToolShedRepository.table.c.name == request.name)
-        if request.owner is not None:
-            clause_list.append(ToolShedRepository.table.c.owner == request.owner)
-        if request.changeset is not None:
-            clause_list.append(ToolShedRepository.table.c.changeset_revision == request.changeset)
-        if request.deleted is not None:
-            clause_list.append(ToolShedRepository.table.c.deleted == request.deleted)
-        if request.uninstalled is not None:
-            clause_list.append(ToolShedRepository.table.c.uninstalled == request.uninstalled)
-        query = (
-            self._install_model_context.query(ToolShedRepository)
-            .order_by(ToolShedRepository.table.c.name)
-            .order_by(cast(ToolShedRepository.ctx_rev, Integer).desc())
+        repositories = self._get_tool_shed_repositories(
+            name=request.name,
+            owner=request.owner,
+            changeset_revision=request.changeset,
+            deleted=request.deleted,
+            uninstalled=request.uninstalled,
         )
-        if len(clause_list) > 0:
-            query = query.filter(and_(*clause_list))
         index = []
-        for repository in query.all():
+        for repository in repositories:
             index.append(self._show(repository))
         return index
 
     def show(self, repository_id: DecodedDatabaseIdField) -> InstalledToolShedRepository:
-        tool_shed_repository = get_tool_shed_repository_by_decoded_id(self._install_model_context, int(repository_id))
+        tool_shed_repository = self._install_model_context.get(ToolShedRepository, repository_id)
         return self._show(tool_shed_repository)
 
     def check_for_updates(self, repository_id: Optional[int]) -> CheckForUpdatesResponse:
@@ -81,3 +67,13 @@ def _show(self, tool_shed_repository: ToolShedRepository) -> InstalledToolShedRe
         tool_shed_repository_dict["error_message"] = tool_shed_repository.error_message or ""
         tool_shed_repository_dict["url"] = url_for("tool_shed_repositories", id=encoded_id)
         return InstalledToolShedRepository(**tool_shed_repository_dict)
+
+    def _get_tool_shed_repositories(self, **kwd):
+        stmt = select(ToolShedRepository)
+        for key, value in kwd.items():
+            if value is not None:
+                column = ToolShedRepository.table.c[key]
+                stmt = stmt.filter(column == value)
+        stmt = stmt.order_by(ToolShedRepository.name).order_by(cast(ToolShedRepository.ctx_rev, Integer).desc())
+        session = self._install_model_context
+        return session.scalars(stmt).all()

lib/galaxy/webapps/galaxy/services/tools.py

@@ -23,7 +23,10 @@
     ProvidesUserContext,
 )
 from galaxy.managers.histories import HistoryManager
-from galaxy.model import PostJobAction
+from galaxy.model import (
+    LibraryDatasetDatasetAssociation,
+    PostJobAction,
+)
 from galaxy.model.base import transaction
 from galaxy.schema.fetch_data import (
     FetchDataFormPayload,
@@ -277,7 +280,7 @@ def _patch_library_inputs(self, trans: ProvidesHistoryContext, inputs, target_hi
 
     def _patch_library_dataset(self, trans: ProvidesHistoryContext, v, target_history):
         if isinstance(v, dict) and "id" in v and v.get("src") == "ldda":
-            ldda = trans.sa_session.query(trans.app.model.LibraryDatasetDatasetAssociation).get(self.decode_id(v["id"]))
+            ldda = trans.sa_session.get(LibraryDatasetDatasetAssociation, self.decode_id(v["id"]))
             if trans.user_is_admin or trans.app.security_agent.can_access_dataset(
                 trans.get_current_user_roles(), ldda.dataset
             ):

lib/galaxy/webapps/galaxy/services/users.py

@@ -7,6 +7,7 @@
 from sqlalchemy import (
     false,
     or_,
+    select,
     true,
 )
 
@@ -204,30 +205,30 @@ def get_index(
         f_any: Optional[str],
     ) -> List[Union[UserModel, LimitedUserModel]]:
         rval = []
-        query = trans.sa_session.query(User)
+        stmt = select(User)
 
         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
@@ -239,8 +240,8 @@ def get_index(
             ):
                 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