diff --git a/lib/galaxy/managers/base.py b/lib/galaxy/managers/base.py index 91a967c62c5..f3abe35e7ce 100644 --- a/lib/galaxy/managers/base.py +++ b/lib/galaxy/managers/base.py @@ -323,17 +323,6 @@ class ModelManager(Generic[U]): except sqlalchemy.orm.exc.MultipleResultsFound: raise exceptions.InconsistentDatabase(f"found more than one {self.model_class.__name__}") - def _one_or_none(self, query): - """ - Return the object if found, None if it's not. - - :raises exceptions.InconsistentDatabase: if more than one model is found - """ - try: - return self._one_with_recast_errors(query) - except exceptions.ObjectNotFound: - return None - # NOTE: at this layer, all ids are expected to be decoded and in int form def by_id(self, id: int) -> Query: """ diff --git a/lib/galaxy/managers/tools.py b/lib/galaxy/managers/tools.py index 1b05ff43b46..639fd058347 100644 --- a/lib/galaxy/managers/tools.py +++ b/lib/galaxy/managers/tools.py @@ -6,13 +6,17 @@ from typing import ( ) from uuid import UUID -from sqlalchemy import sql +from sqlalchemy import ( + select, + sql, +) from galaxy import ( exceptions, model, ) from galaxy.exceptions import DuplicatedIdentifierException +from galaxy.model import DynamicTool from galaxy.tool_util.cwl import tool_proxy from .base import ( ModelManager, @@ -32,16 +36,16 @@ class DynamicToolManager(ModelManager): model_class = model.DynamicTool def get_tool_by_uuid(self, uuid: Optional[Union[UUID, str]]): - dynamic_tool = self._one_or_none(self.query().filter(self.model_class.uuid == uuid)) - return dynamic_tool + stmt = select(DynamicTool).where(DynamicTool.uuid == uuid) + return self.session().scalars(stmt).one_or_none() def get_tool_by_tool_id(self, tool_id): - dynamic_tool = self._one_or_none(self.query().filter(self.model_class.tool_id == tool_id)) - return dynamic_tool + stmt = select(DynamicTool).where(DynamicTool.tool_id == tool_id) + return self.session().scalars(stmt).one_or_none() def get_tool_by_id(self, object_id): - dynamic_tool = self._one_or_none(self.query().filter(self.model_class.id == object_id)) - return dynamic_tool + stmt = select(DynamicTool).where(DynamicTool.id == object_id) + return self.session().scalars(stmt).one_or_none() def create_tool(self, trans, tool_payload, allow_load=True): if not getattr(self.app.config, "enable_beta_tool_formats", False): @@ -112,7 +116,8 @@ class DynamicToolManager(ModelManager): return dynamic_tool def list_tools(self, active=True): - return self.query().filter(self.model_class.active == active) + stmt = select(DynamicTool).where(DynamicTool.active == active) + return self.session().scalars(stmt) def deactivate(self, dynamic_tool): self.update(dynamic_tool, {"active": False})