Minor fix to restore admin-style dynamic tools

lib/galaxy/managers/tools.py

@@ -16,7 +16,9 @@
     model,
 )
 from galaxy.exceptions import DuplicatedIdentifierException
+from galaxy.managers.context import ProvidesUserContext
 from galaxy.model import DynamicTool
+from galaxy.schema.tools import DynamicToolPayload
 from galaxy.tool_util.cwl import tool_proxy
 from .base import (
     ModelManager,
@@ -47,7 +49,7 @@ def get_tool_by_id(self, object_id):
         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):
+    def create_tool(self, trans: ProvidesUserContext, tool_payload: DynamicToolPayload, allow_load=True):
         if not getattr(self.app.config, "enable_beta_tool_formats", False):
             raise exceptions.ConfigDoesNotAllowException(
                 "Set 'enable_beta_tool_formats' in Galaxy config to create dynamic tools."
@@ -63,33 +65,31 @@ def create_tool(self, trans, tool_payload, allow_load=True):
             dynamic_tool = self.get_tool_by_uuid(uuid_str)
             if dynamic_tool:
                 if not allow_load:
-                    raise DuplicatedIdentifierException(dynamic_tool.id)
+                    raise DuplicatedIdentifierException(
+                        f"Attempted to create dynamic tool with duplicate UUID '{uuid_str}'"
+                    )
                 assert dynamic_tool.uuid == uuid
         if not dynamic_tool:
-            src = tool_payload.src
-            is_path = src == "from_path"
-
-            if is_path:
-                tool_format, representation, _ = artifact_class(None, tool_payload)
+            if tool_payload.src == "from_path":
+                tool_format, representation, _ = artifact_class(None, tool_payload.model_dump())
             else:
-                assert src == "representation"
-                representation = tool_payload.representation.dict(by_alias=True, exclude_unset=True)
+                representation = tool_payload.representation.model_dump(by_alias=True, exclude_unset=True)
                 if not representation:
                     raise exceptions.ObjectAttributeMissingException("A tool 'representation' is required.")
 
                 tool_format = representation.get("class")
                 if not tool_format:
                     raise exceptions.ObjectAttributeMissingException("Current tool representations require 'class'.")
 
-            tool_path = tool_payload.path
             tool_directory = tool_payload.tool_directory
+            tool_path = tool_payload.path if tool_payload.src == "from_path" else None
             if tool_format in ("GalaxyTool", "GalaxyUserTool"):
                 tool_id = representation.get("id")
                 if not tool_id:
                     tool_id = str(uuid)
             elif tool_format in ("CommandLineTool", "ExpressionTool"):
                 # CWL tools
-                if is_path:
+                if tool_path:
                     proxy = tool_proxy(tool_path=tool_path, uuid=uuid)
                 else:
                     # Build a tool proxy so that we can convert to the persistable

lib/galaxy/schema/tools.py

@@ -0,0 +1,30 @@
+from typing import (
+    Literal,
+    Optional,
+    Union,
+)
+
+from pydantic import BaseModel
+
+from galaxy.tool_util.models import DynamicToolSources
+
+
+class BaseDynamicToolCreatePayload(BaseModel):
+    allow_load: bool = True
+    uuid: Optional[str] = None
+    active: Optional[bool] = None
+    hidden: Optional[bool] = None
+    tool_directory: Optional[str] = None
+
+
+class DynamicToolCreatePayload(BaseDynamicToolCreatePayload):
+    src: Literal["representation"] = "representation"
+    representation: DynamicToolSources
+
+
+class PathBasedDynamicToolCreatePayload(BaseDynamicToolCreatePayload):
+    src: Literal["from_path"]
+    path: str
+
+
+DynamicToolPayload = Union[DynamicToolCreatePayload, PathBasedDynamicToolCreatePayload]

lib/galaxy/tool_util/models.py

@@ -18,6 +18,7 @@
     BaseModel,
     ConfigDict,
     Field,
+    model_validator,
     RootModel,
 )
 from typing_extensions import (
@@ -47,15 +48,22 @@
 from .verify.assertion_models import assertions
 
 
-class UserToolSource(BaseModel):
-    class_: Annotated[Literal["GalaxyUserTool"], Field(alias="class")]
-    name: str
-    container: str
-    version: str
+def normalize_dict(values, keys: List[str]):
+    for key in keys:
+        items = values.get(key)
+        if isinstance(items, dict):  # dict-of-dicts format
+            # Transform dict-of-dicts to list-of-dicts
+            values[key] = [{"name": k, **v} for k, v in items.items()]
+
+
+class ToolSourceBase(BaseModel):
+    id: Optional[str] = None
+    name: Optional[str] = None
+    version: Optional[str] = "1.0"
     description: Optional[str] = None
-    shell_command: str
-    inputs: List[GalaxyParameterT]
-    outputs: List[IncomingToolOutput]
+    container: Optional[str] = None
+    inputs: List[GalaxyParameterT] = []
+    outputs: List[IncomingToolOutput] = []
     citations: Optional[List[Citation]] = None
     license: Optional[str] = None
     profile: Optional[str] = None
@@ -64,6 +72,28 @@ class UserToolSource(BaseModel):
     xrefs: Optional[List[XrefDict]] = None
     help: Optional[HelpContent] = None
 
+    @model_validator(mode="before")
+    @classmethod
+    def normalize_items(cls, values):
+        if isinstance(values, dict):
+            normalize_dict(values, ["inputs", "outputs"])
+        return values
+
+
+class UserToolSource(ToolSourceBase):
+    class_: Annotated[Literal["GalaxyUserTool"], Field(alias="class")]
+    name: str
+    shell_command: str
+    container: str
+
+
+class AdminToolSource(ToolSourceBase):
+    class_: Annotated[Literal["GalaxyTool"], Field(alias="class")]
+    command: str
+
+
+DynamicToolSources = Annotated[Union[UserToolSource, AdminToolSource], Field(discriminator="class_")]
+
 
 class ParsedTool(BaseModel):
     id: str

lib/galaxy/webapps/galaxy/api/dynamic_tools.py

@@ -1,34 +1,16 @@
 import logging
-from typing import (
-    Literal,
-    Optional,
-)
-
-from pydantic.main import BaseModel
 
 from galaxy.exceptions import ObjectNotFound
 from galaxy.managers.context import ProvidesUserContext
 from galaxy.managers.tools import DynamicToolManager
 from galaxy.schema.fields import DecodedDatabaseIdField
-from galaxy.tool_util.models import UserToolSource
+from galaxy.schema.tools import DynamicToolPayload
 from . import (
     depends,
     DependsOnTrans,
     Router,
 )
 
-
-class DynamicToolCreatePayload(BaseModel):
-    allow_load: bool = True
-    uuid: Optional[str] = None
-    src: Literal["representation", "path"] = "representation"
-    representation: UserToolSource
-    active: Optional[bool] = None
-    hidden: Optional[bool] = None
-    path: Optional[str] = None
-    tool_directory: Optional[str] = None
-
-
 log = logging.getLogger(__name__)
 
 router = Router(tags=["dynamic_tools"])
@@ -50,7 +32,7 @@ def show(self, dynamic_tool_id: DecodedDatabaseIdField):
         return dynamic_tool.to_dict()
 
     @router.post("/api/dynamic_tools", require_admin=True)
-    def create(self, payload: DynamicToolCreatePayload, trans: ProvidesUserContext = DependsOnTrans):
+    def create(self, payload: DynamicToolPayload, trans: ProvidesUserContext = DependsOnTrans):
         dynamic_tool = self.dynamic_tools_manager.create_tool(trans, payload, allow_load=payload.allow_load)
         return dynamic_tool.to_dict()
 

lib/galaxy/workflow/modules.py

@@ -53,6 +53,7 @@
     InvocationFailureWhenNotBoolean,
     InvocationFailureWorkflowParameterInvalid,
 )
+from galaxy.schema.tools import DynamicToolCreatePayload
 from galaxy.tool_util.cwl.util import set_basename_and_derived_properties
 from galaxy.tool_util.parser import get_input_source
 from galaxy.tool_util.parser.output_objects import ToolExpressionOutput
@@ -1887,9 +1888,7 @@ def from_dict(Class, trans, d, **kwds):
         if tool_id is None and tool_uuid is None:
             tool_representation = d.get("tool_representation")
             if tool_representation:
-                create_request = {
-                    "representation": tool_representation,
-                }
+                create_request = DynamicToolCreatePayload(src="representation", representation=tool_representation)
                 if not trans.user_is_admin:
                     raise exceptions.AdminRequiredException("Only admin users can create tools dynamically.")
                 dynamic_tool = trans.app.dynamic_tool_manager.create_tool(trans, create_request, allow_load=False)

lib/galaxy_test/api/embed_test_1_tool.gxtool.yml

@@ -3,3 +3,4 @@ command: echo 'hello world 2' > $output1
 outputs:
   output1:
     format: txt
+    type: data

lib/galaxy_test/api/test_tools.py

@@ -43,40 +43,17 @@
     "version": "1.0.0",
     "command": "echo 'Hello World' > $output1",
     "inputs": [],
-    "outputs": dict(
-        output1=dict(format="txt"),
-    ),
+    "outputs": {"output1": {"format": "txt", "type": "data"}},
 }
 MINIMAL_TOOL_NO_ID = {
     "name": "Minimal Tool",
     "class": "GalaxyTool",
     "version": "1.0.0",
     "command": "echo 'Hello World 2' > $output1",
     "inputs": [],
-    "outputs": dict(
-        output1=dict(format="txt"),
-    ),
-}
-TOOL_WITH_BASE_COMMAND = {
-    "name": "Base command tool",
-    "class": "GalaxyTool",
-    "version": "1.0.0",
-    "base_command": "cat",
-    "arguments": ["$(inputs.input.path)", ">", "output.fastq"],
-    "inputs": [
-        {
-            "type": "data",
-            "name": "input",
-        }
-    ],
-    "outputs": {
-        "output": {
-            "type": "data",
-            "from_work_dir": "output.fastq",
-            "name": "output",
-        }
-    },
+    "outputs": {"output1": {"format": "txt", "type": "data"}},
 }
+
 TOOL_WITH_SHELL_COMMAND = {
     "name": "Base command tool",
     "class": "GalaxyUserTool",
@@ -1422,22 +1399,44 @@ def test_dynamic_tool_no_id(self):
         output_content = self.dataset_populator.get_history_dataset_content(history_id)
         assert output_content == "Hello World 2\n"
 
-    def test_dynamic_tool_base_command(self):
-        tool_response = self.dataset_populator.create_tool(TOOL_WITH_BASE_COMMAND)
-        self._assert_has_keys(tool_response, "uuid")
-
-        # Run tool.
-        history_id = self.dataset_populator.new_history()
-        dataset = self.dataset_populator.new_dataset(history_id=history_id, content="abc")
-        self._run(
-            history_id=history_id,
-            tool_uuid=tool_response["uuid"],
-            inputs={"input": {"src": "hda", "id": dataset["id"]}},
-        )
-
-        self.dataset_populator.wait_for_history(history_id, assert_ok=True)
-        output_content = self.dataset_populator.get_history_dataset_content(history_id)
-        assert output_content == "abc\n"
+    # This works except I don't want to add it to the schema right now,
+    # since I think the shell_command is what we'lll go with (at least initially)
+    # def test_dynamic_tool_base_command(self):
+    #     TOOL_WITH_BASE_COMMAND = {
+    #         "name": "Base command tool",
+    #         "class": "GalaxyTool",
+    #         "version": "1.0.0",
+    #         "base_command": "cat",
+    #         "arguments": ["$(inputs.input.path)", ">", "output.fastq"],
+    #         "inputs": [
+    #             {
+    #                 "type": "data",
+    #                 "name": "input",
+    #             }
+    #         ],
+    #         "outputs": {
+    #             "output": {
+    #                 "type": "data",
+    #                 "from_work_dir": "output.fastq",
+    #                 "name": "output",
+    #             }
+    #         },
+    #     }
+    #     tool_response = self.dataset_populator.create_tool(TOOL_WITH_BASE_COMMAND)
+    #     self._assert_has_keys(tool_response, "uuid")
+
+    #     # Run tool.
+    #     history_id = self.dataset_populator.new_history()
+    #     dataset = self.dataset_populator.new_dataset(history_id=history_id, content="abc")
+    #     self._run(
+    #         history_id=history_id,
+    #         tool_uuid=tool_response["uuid"],
+    #         inputs={"input": {"src": "hda", "id": dataset["id"]}},
+    #     )
+
+    #     self.dataset_populator.wait_for_history(history_id, assert_ok=True)
+    #     output_content = self.dataset_populator.get_history_dataset_content(history_id)
+    #     assert output_content == "abc\n"
 
     def test_dynamic_tool_shell_command(self):
         tool_response = self.dataset_populator.create_tool(TOOL_WITH_SHELL_COMMAND)

lib/galaxy_test/api/test_workflows.py

@@ -1097,6 +1097,7 @@ def test_import_export_dynamic(self):
       outputs:
         output1:
           format: txt
+          type: data
   - tool_id: cat1
     state:
       input1:
@@ -7792,6 +7793,7 @@ def test_import_export_dynamic_tools(self, history_id):
       outputs:
         output1:
           format: txt
+          type: data
   - tool_id: cat1
     state:
       input1:

lib/galaxy_test/api/test_workflows_from_yaml.py

@@ -323,6 +323,7 @@ def test_workflow_embed_tool(self):
       outputs:
         output1:
           format: txt
+          type: data
   - tool_id: cat1
     state:
       input1:

lib/galaxy_test/base/data/minimal_tool_no_id.json

@@ -6,7 +6,8 @@
     "inputs": [],
     "outputs": {
         "output1": {
-            "format": "txt"
+            "format": "txt",
+            "type": "data"
         }
     }
 }

lib/galaxy_test/base/populators.py

@@ -2451,15 +2451,15 @@ def import_workflow(self, workflow, **kwds) -> Dict[str, Any]:
         }
         data.update(**kwds)
         upload_response = self._post("workflows", data=data)
-        assert upload_response.status_code == 200, upload_response.content
+        assert upload_response.status_code == 200, upload_response.text
         return upload_response.json()
 
     def import_tool(self, tool) -> Dict[str, Any]:
         """Import a workflow via POST /api/workflows or
         comparable interface into Galaxy.
         """
         upload_response = self._import_tool_response(tool)
-        assert upload_response.status_code == 200, upload_response
+        assert upload_response.status_code == 200, upload_response.text
         return upload_response.json()
 
     def build_module(self, step_type: str, content_id: Optional[str] = None, inputs: Optional[Dict[str, Any]] = None):
@@ -2470,9 +2470,8 @@ def build_module(self, step_type: str, content_id: Optional[str] = None, inputs:
 
     def _import_tool_response(self, tool) -> Response:
         using_requirement("admin")
-        tool_str = json.dumps(tool, indent=4)
-        data = {"representation": tool_str}
-        upload_response = self._post("dynamic_tools", data=data, admin=True)
+        data = {"representation": tool}
+        upload_response = self._post("dynamic_tools", data=data, admin=True, json=True)
         return upload_response
 
     def scaling_workflow_yaml(self, **kwd):