Add tsr_repo.get_filtered, use in services.tool_shed_repository

lib/galaxy/model/repositories/tool_shed_repository.py

@@ -1,3 +1,10 @@
+from sqlalchemy import (
+    and_,
+    cast,
+    Integer,
+    select,
+)
+
 from galaxy.model.repositories import (
     ModelRepository,
     SessionType,
@@ -8,3 +15,25 @@
 class ToolShedRepositoryRepository(ModelRepository):
     def __init__(self, session: SessionType):
         super().__init__(session, ToolShedRepository)
+
+    def get_filtered(self, name, owner, changeset, deleted, uninstalled):
+        clause_list = []
+        if name is not None:
+            clause_list.append(ToolShedRepository.name == name)
+        if owner is not None:
+            clause_list.append(ToolShedRepository.owner == owner)
+        if changeset is not None:
+            clause_list.append(ToolShedRepository.changeset_revision == changeset)
+        if deleted is not None:
+            clause_list.append(ToolShedRepository.deleted == deleted)
+        if uninstalled is not None:
+            clause_list.append(ToolShedRepository.uninstalled == uninstalled)
+
+        stmt = (
+            select(ToolShedRepository)
+            .order_by(ToolShedRepository.name)
+            .order_by(cast(ToolShedRepository.ctx_rev, Integer).desc())
+        )
+        if len(clause_list) > 0:
+            stmt = stmt.filter(and_(*clause_list))
+        return self.session.scalars(stmt).all()  # type:ignore[union-attr]

lib/galaxy/tool_shed/util/repository_util.py

@@ -26,6 +26,7 @@
     web,
 )
 from galaxy.model.base import transaction
+from galaxy.model.repositories.tool_shed_repository import ToolShedRepositoryRepository as tsr_repo
 from galaxy.model.scoped_session import install_model_scoped_session
 from galaxy.model.tool_shed_install import ToolShedRepository
 from galaxy.tool_shed.util import basic_util
@@ -69,7 +70,7 @@ def check_for_updates(
             message += "Unable to retrieve status from the tool shed for the following repositories:\n"
             message += ", ".join(repository_names_not_updated)
     else:
-        repository = get_tool_shed_repository_by_decoded_id(install_model_context, repository_id)
+        repository = tsr_repo(install_model_context).get(repository_id)
         ok, updated = _check_or_update_tool_shed_status_for_installed_repository(
             tool_shed_registry, install_model_context, repository
         )
@@ -632,15 +633,7 @@ def get_tool_shed_from_clone_url(repository_clone_url):
 def get_tool_shed_repository_by_id(app, repository_id) -> ToolShedRepository:
     """Return a tool shed repository database record defined by the id."""
     # This method is used only in Galaxy, not the tool shed.
-    return get_tool_shed_repository_by_decoded_id(app.install_model.context, app.security.decode_id(repository_id))
-
-
-def get_tool_shed_repository_by_decoded_id(
-    install_model_context: install_model_scoped_session, repository_id: int
-) -> ToolShedRepository:
-    return (
-        install_model_context.query(ToolShedRepository).filter(ToolShedRepository.table.c.id == repository_id).first()
-    )
+    return tsr_repo(app.install_model_context).get(app.security.decode_id(repository_id))
 
 
 def get_tool_shed_status_for(tool_shed_registry: Registry, repository: ToolShedRepository):

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

@@ -4,24 +4,16 @@
 )
 
 from pydantic import BaseModel
-from sqlalchemy import (
-    and_,
-    cast,
-    Integer,
-    select,
-)
 
+from galaxy.model.repositories.tool_shed_repository import ToolShedRepositoryRepository as tsr_repo
 from galaxy.model.scoped_session import install_model_scoped_session
 from galaxy.model.tool_shed_install import ToolShedRepository
 from galaxy.schema.fields import DecodedDatabaseIdField
 from galaxy.schema.schema import (
     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
 
@@ -42,33 +34,23 @@ def __init__(
     ):
         self._install_model_context = install_model_context
         self._tool_shed_registry = tool_shed_registry
+        self._tsr_repo = tsr_repo(self._install_model_context)
 
     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)
-        stmt = (
-            select(ToolShedRepository)
-            .order_by(ToolShedRepository.table.c.name)
-            .order_by(cast(ToolShedRepository.ctx_rev, Integer).desc())
+        repositories = self._tsr_repo.get_filtered(
+            name=request.name,
+            owner=request.owner,
+            changeset=request.changeset,
+            deleted=request.deleted,
+            uninstalled=request.uninstalled,
         )
-        if len(clause_list) > 0:
-            stmt = stmt.filter(and_(*clause_list))
         index = []
-        for repository in self._install_model_context.scalars(stmt).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._tsr_repo.get(repository_id)
         return self._show(tool_shed_repository)
 
     def check_for_updates(self, repository_id: Optional[int]) -> CheckForUpdatesResponse: