Improve type annotation of workflow management methods

lib/galaxy/model/__init__.py

@@ -417,11 +417,11 @@ def set_datatypes_registry(d_registry):
     _datatypes_registry = d_registry
 
 
-class HasTags:
+class HasTags(Dictifiable):
     dict_collection_visible_keys = ["tags"]
     dict_element_visible_keys = ["tags"]
 
-    def to_dict(self, *args, **kwargs):
+    def to_dict(self, *args, **kwargs) -> dict[str, Any]:
         rval = super().to_dict(*args, **kwargs)
         rval["tags"] = self.make_tag_string_list()
         return rval
@@ -3508,7 +3508,7 @@ def prune(cls, sa_session):
             session.execute(q)
 
 
-class History(Base, HasTags, Dictifiable, UsesAnnotations, HasName, Serializable, UsesCreateAndUpdateTime):
+class History(Base, HasTags, UsesAnnotations, HasName, Serializable, UsesCreateAndUpdateTime):
     __tablename__ = "history"
     __table_args__ = (Index("ix_history_slug", "slug", mysql_length=200),)
 
@@ -5875,7 +5875,7 @@ def _handle_serialize_files(self, id_encoder, serialization_options, rval):
             rval["file_metadata"] = file_metadata
 
 
-class HistoryDatasetAssociation(DatasetInstance, HasTags, Dictifiable, UsesAnnotations, HasName, Serializable):
+class HistoryDatasetAssociation(DatasetInstance, HasTags, UsesAnnotations, HasName, Serializable):
     """
     Resource class that creates a relation between a dataset and a user history.
     """
@@ -7830,7 +7830,6 @@ class HistoryDatasetCollectionAssociation(
     Base,
     DatasetCollectionInstance,
     HasTags,
-    Dictifiable,
     UsesAnnotations,
     Serializable,
 ):
@@ -8582,7 +8581,7 @@ def __init__(self, galaxy_session, history):
         self.history = history
 
 
-class StoredWorkflow(Base, HasTags, Dictifiable, RepresentById, UsesCreateAndUpdateTime):
+class StoredWorkflow(Base, HasTags, RepresentById, UsesCreateAndUpdateTime):
     """
     StoredWorkflow represents the root node of a tree of objects that compose a workflow, including workflow revisions, steps, and subworkflows.
     It is responsible for the metadata associated with a workflow including owner, name, published, and create/update time.
@@ -8703,7 +8702,7 @@ def __init__(
         self.workflows = listify(workflow)
         self.hidden = hidden
 
-    def get_internal_version(self, version: Optional[int] = None):
+    def get_internal_version(self, version: Optional[int] = None) -> "Workflow":
         if version is None:
             return self.latest_workflow
         if len(self.workflows) <= version:
@@ -8756,7 +8755,7 @@ def invocation_counts(self) -> InvocationsStateCounts:
         rows_as_dict = dict(r for r in rows if r[0] is not None)  # type: ignore[arg-type, var-annotated]
         return InvocationsStateCounts(rows_as_dict)
 
-    def to_dict(self, view="collection", value_mapper=None):
+    def to_dict(self, view="collection", value_mapper=None) -> dict[str, Any]:
         rval = super().to_dict(view=view, value_mapper=value_mapper)
         rval["latest_workflow_uuid"] = (lambda uuid: str(uuid) if self.latest_workflow.uuid else None)(
             self.latest_workflow.uuid
@@ -11520,7 +11519,7 @@ def create_social_auth(cls, user, uid, provider):
         return instance
 
 
-class Page(Base, HasTags, Dictifiable, RepresentById, UsesCreateAndUpdateTime):
+class Page(Base, HasTags, RepresentById, UsesCreateAndUpdateTime):
     __tablename__ = "page"
     __table_args__ = (Index("ix_page_slug", "slug", mysql_length=200),)
 
@@ -11641,7 +11640,7 @@ class PageUserShareAssociation(Base, UserShareAssociation):
     page: Mapped["Page"] = relationship(back_populates="users_shared_with")
 
 
-class Visualization(Base, HasTags, Dictifiable, RepresentById, UsesCreateAndUpdateTime):
+class Visualization(Base, HasTags, RepresentById, UsesCreateAndUpdateTime):
     __tablename__ = "visualization"
     __table_args__ = (
         Index("ix_visualization_dbkey", "dbkey", mysql_length=200),

lib/galaxy/model/store/__init__.py

@@ -2579,7 +2579,7 @@ def record_associated_jobs(obj):
             if not self.app:
                 raise Exception(f"Missing self.app in {self}.")
             self.app.workflow_contents_manager.store_workflow_artifacts(
-                workflows_directory, workflow_key, workflow, user=history.user, history=history
+                workflows_directory, workflow_key, workflow, history=history, user=history.user
             )
             invocations_attrs.append(invocation_attrs)
 

lib/galaxy/webapps/base/controller.py

@@ -6,6 +6,7 @@
 from collections.abc import Callable
 from typing import (
     Any,
+    TYPE_CHECKING,
 )
 
 from webob.exc import (
@@ -59,6 +60,9 @@
 )
 from galaxy.workflow.modules import WorkflowModuleInjector
 
+if TYPE_CHECKING:
+    from galaxy.structured_app import StructuredApp
+
 log = logging.getLogger(__name__)
 
 # States for passing messages
@@ -582,6 +586,7 @@ def _get_highest_priority_msg(self, message_list):
 class UsesStoredWorkflowMixin(SharableItemSecurityMixin, UsesAnnotations):
     """Mixin for controllers that use StoredWorkflow objects."""
 
+    app: "StructuredApp"
     slug_builder = SlugBuilder()
 
     def get_stored_workflow(self, trans, id, check_ownership=True, check_accessible=False):
@@ -637,7 +642,7 @@ def _import_shared_workflow(self, trans, stored: StoredWorkflow):
         session.commit()
         return imported_stored
 
-    def _workflow_to_dict(self, trans, stored):
+    def _workflow_to_dict(self, trans, stored: StoredWorkflow) -> dict[str, Any]:
         """
         Converts a workflow to a dict of attributes suitable for exporting.
         """

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

@@ -343,6 +343,11 @@ def workflow_dict(self, trans: GalaxyWebTransaction, workflow_id, **kwd):
         style = kwd.get("style", "export")
         download_format = kwd.get("format")
         version = kwd.get("version")
+        if version is not None:
+            try:
+                version = int(version)
+            except ValueError:
+                raise exceptions.RequestParameterInvalidException("Invalid version specified.")
         history = None
         if history_id := kwd.get("history_id"):
             history = self.history_manager.get_accessible(

lib/galaxy/webapps/galaxy/controllers/workflow.py

@@ -24,6 +24,7 @@
 
 
 class WorkflowController(BaseUIController, SharableMixin, UsesStoredWorkflowMixin, UsesItemRatings):
+    app: StructuredApp
     history_manager: HistoryManager = depends(HistoryManager)
     slug_builder = SlugBuilder()
 

lib/galaxy/workflow/modules.py

@@ -777,7 +777,7 @@ def get_modules(self):
         return self._modules
 
     @property
-    def version_changes(self):
+    def version_changes(self) -> list[str]:
         version_changes = []
         for m in self.get_modules():
             if hasattr(m, "version_changes"):

lib/galaxy_test/api/test_workflows.py

@@ -1544,7 +1544,9 @@ def test_trs_import(self):
             "trs_tool_id": "#workflow/github.com/jmchilton/galaxy-workflow-dockstore-example-1/mycoolworkflow",
             "trs_version_id": "master",
         }
-        workflow_id = self._post("workflows", data=trs_payload).json()["id"]
+        response = self._post("workflows", data=trs_payload)
+        response.raise_for_status()
+        workflow_id = response.json()["id"]
         original_workflow = self._download_workflow(workflow_id)
         assert "Test Workflow" in original_workflow["name"]
         assert original_workflow.get("source_metadata").get("trs_tool_id") == trs_payload["trs_tool_id"]
@@ -1571,7 +1573,9 @@ def test_trs_import_from_dockstore_trs_url(self):
             "%23workflow%2Fgithub.com%2Fjmchilton%2Fgalaxy-workflow-dockstore-example-1%2Fmycoolworkflow/"
             "versions/master",
         }
-        workflow_id = self._post("workflows", data=trs_payload).json()["id"]
+        response = self._post("workflows", data=trs_payload)
+        response.raise_for_status()
+        workflow_id = response.json()["id"]
         original_workflow = self._download_workflow(workflow_id)
         assert "Test Workflow" in original_workflow["name"]
         assert (
@@ -1604,7 +1608,9 @@ def test_trs_import_from_workflowhub_trs_url(self):
             "archive_source": "trs_tool",
             "trs_url": "https://workflowhub.eu/ga4gh/trs/v2/tools/109/versions/5",
         }
-        workflow_id = self._post("workflows", data=trs_payload).json()["id"]
+        response = self._post("workflows", data=trs_payload)
+        response.raise_for_status()
+        workflow_id = response.json()["id"]
         original_workflow = self._download_workflow(workflow_id)
         assert "COVID-19: variation analysis reporting" in original_workflow["name"]
         assert original_workflow.get("source_metadata").get("trs_tool_id") == "109"

mypy.ini

@@ -6,7 +6,7 @@ ignore_missing_imports = True
 check_untyped_defs = True
 exclude = (?x)(
         ^build/
-        | ^test/functional/
+        | .*test/functional/tools/cwl_tools/
         | .*tool_shed/test/test_data/repos/
     )
 pretty = True