Move method into User model class

Author: jdavcs

lib/galaxy/managers/workflows.py

@@ -27,16 +27,13 @@
     and_,
     desc,
     false,
-    func,
     or_,
-    select,
     true,
 )
 from sqlalchemy.orm import (
     aliased,
     joinedload,
     Query,
-    Session,
     subqueryload,
 )
 
@@ -53,10 +50,7 @@
 from galaxy.managers.base import decode_id
 from galaxy.managers.context import ProvidesUserContext
 from galaxy.managers.executables import artifact_class
-from galaxy.model import (
-    StoredWorkflow,
-    StoredWorkflowUserShareAssociation,
-)
+from galaxy.model import StoredWorkflow
 from galaxy.model.base import transaction
 from galaxy.model.index_filter_util import (
     append_user_filter,
@@ -1978,9 +1972,3 @@ def import_workflow(self, workflow, **kwds):
         raise NotImplementedError(
             "Direct format 2 import of nested workflows is not yet implemented, use bioblend client."
         )
-
-
-def count_stored_workflow_user_assocs(session: Session, user, stored_workflow) -> int:
-    stmt = select(StoredWorkflowUserShareAssociation).filter_by(user=user, stored_workflow=stored_workflow)
-    stmt = select(func.count()).select_from(stmt)
-    return session.scalar(stmt)

lib/galaxy/model/__init__.py

@@ -1169,6 +1169,12 @@ def quota_source_usage_for(self, quota_source_label: Optional[str]) -> Optional[
                 return quota_source_usage
         return None
 
+    def count_stored_workflow_user_assocs(self, stored_workflow) -> int:
+        stmt = select(StoredWorkflowUserShareAssociation).filter_by(user=self, stored_workflow=stored_workflow)
+        stmt = select(func.count()).select_from(stmt)
+        session = object_session(self)
+        return session.scalar(stmt)
+
 
 class PasswordResetToken(Base):
     __tablename__ = "password_reset_token"

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

@@ -40,7 +40,6 @@
     invocation_job_source_iter,
 )
 from galaxy.managers.workflows import (
-    count_stored_workflow_user_assocs,
     MissingToolsException,
     RefactorRequest,
     WorkflowCreateOptions,
@@ -193,7 +192,7 @@ 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:
-            wf_count = count_stored_workflow_user_assocs(trans.sa_session, trans.user, stored_workflow)
+            wf_count = trans.user.count_stored_workflow_user_assocs(stored_workflow)
             if wf_count == 0:
                 message = "Workflow is neither importable, nor owned by or shared with current user"
                 raise exceptions.ItemAccessibilityException(message)