Add method to WorkflowRepository, api.workflows

count_stored_workflow_user_assocs

lib/galaxy/model/repositories/workflow.py

@@ -1,6 +1,14 @@
 from typing import cast
 
-from galaxy.model import Workflow
+from sqlalchemy import (
+    func,
+    select,
+)
+
+from galaxy.model import (
+    StoredWorkflowUserShareAssociation,
+    Workflow,
+)
 from galaxy.model.repositories import (
     BaseRepository,
     MappedType,
@@ -14,3 +22,9 @@ def __init__(self, session: SessionType):
 
     def get(self, primary_key: int) -> Workflow:
         return cast(Workflow, super().get(primary_key))
+
+    def count_stored_workflow_user_assocs(self, user: MappedType, stored_workflow: MappedType) -> int:
+        # type-ignore/SessionlessContext
+        stmt = select(StoredWorkflowUserShareAssociation).filter_by(user=user, stored_workflow=stored_workflow)
+        stmt = select(func.count()).select_from(stmt)
+        return self.session.scalar(stmt)  # type:ignore[union-attr]

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

@@ -23,10 +23,6 @@
 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 (
@@ -52,6 +48,7 @@
 from galaxy.model.base import transaction
 from galaxy.model.item_attrs import UsesAnnotations
 from galaxy.model.repositories.stored_workflow import StoredWorkflowRepository
+from galaxy.model.repositories.workflow import WorkflowRepository
 from galaxy.model.store import BcoExportOptions
 from galaxy.schema.fields import DecodedDatabaseIdField
 from galaxy.schema.invocation import InvocationMessageResponseModel
@@ -198,11 +195,10 @@ 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:
-            stmt = select(model.StoredWorkflowUserShareAssociation).filter_by(
-                user=trans.user, stored_workflow=stored_workflow
+            wf_count = WorkflowRepository(trans.sa_session).count_stored_workflow_user_assocs(
+                trans.user, stored_workflow
             )
-            stmt = select(func.count()).select_from(stmt)
-            if trans.sa_session.scalar(stmt) == 0:
+            if wf_count == 0:
                 message = "Workflow is neither importable, nor owned by or shared with current user"
                 raise exceptions.ItemAccessibilityException(message)
         if kwd.get("legacy", False):