Fix SA2.0 usage in managers.tools; refactor

Refactor: use native SA one_or_none method instead of calling
self._one_or_none >> self._one_with_recast_errors.
Reason: old code duplicated SA's error catching logic while not
providing a useful error message. Using SA's native method achieves the
same result with a more useful error message and without adding custom code.
The _one_or_none method is dropped as it is not used anywhere else.
This commit is contained in:
John Davis
2023-11-09 15:33:46 -05:00
parent 792ac5015a
commit 884e1ed38a
2 changed files with 13 additions and 19 deletions
-11
View File
@@ -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:
"""
+13 -8
View File
@@ -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})