Implement a subset of the Common Workflow Language

.github/workflows/cwl_conformance.yaml

@@ -20,15 +20,17 @@ concurrency:
 jobs:
   test:
     name: Test
-    if: ${{ false }}
     runs-on: ubuntu-latest
     continue-on-error: ${{ startsWith(matrix.marker, 'red') }}
     strategy:
       fail-fast: false
       matrix:
         python-version: ['3.9']
         marker: ['green', 'red and required', 'red and not required']
-        conformance-version: ['cwl_conformance_v1_0'] #, 'cwl_conformance_v1_1', 'cwl_conformance_v1_2']
+        conformance-version: ['cwl_conformance_v1_0', 'cwl_conformance_v1_1', 'cwl_conformance_v1_2']
+        exclude:
+          - marker: red and required
+            conformance-version: cwl_conformance_v1_0
     services:
       postgres:
         image: postgres:17

doc/source/dev/cwl.md

@@ -0,0 +1,24 @@
+CWL import in Galaxy
+====================
+
+What is supported
+-----------------
+
+What is not supported
+---------------------
+
+Some CWL Expressions / Parameter references that do math on `$(resources.cores)`
+or similar will likely not work.
+
+How to enable it?
+-----------------
+
+1. List paths to CWL tools in `tool_conf.xml` .
+2. Set the following in  `galaxy.yml`: 
+
+   ```yaml
+   enable_beta_tool_formats: true
+   enable_beta_workflow_modules: true
+   check_upload_content: false
+   strict_cwl_validation: false
+   ```

lib/galaxy/config/__init__.py

@@ -937,6 +937,9 @@ def _process_config(self, kwargs: Dict[str, Any]) -> None:
             else None
         )
 
+        # TODO: migrate to schema.
+        # Should CWL artifacts be loaded with strict validation enabled.
+        self.strict_cwl_validation = string_as_bool(kwargs.get("strict_cwl_validation", "True"))
         # These are not even beta - just experiments - don't use them unless
         # you want yours tools to be broken in the future.
         self.enable_beta_tool_formats = string_as_bool(kwargs.get("enable_beta_tool_formats", "False"))

lib/galaxy/config/sample/datatypes_conf.xml.sample

@@ -338,7 +338,7 @@
     <datatype extension="tar" auto_compressed_types="gz,bz2" type="galaxy.datatypes.binary:CompressedArchive" subclass="true" display_in_upload="true">
       <converter file="archive_to_directory.xml" target_datatype="directory"/>
     </datatype>
-    <datatype extension="directory" type="galaxy.datatypes.data:Directory"/>
+    <datatype extension="directory" type="galaxy.datatypes.data:Directory" display_in_upload="true"/>
     <datatype extension="bwa_mem2_index" display_in_upload="true" type="galaxy.datatypes.data:Directory" subclass="true"/>
     <datatype extension="zarr" type="galaxy.datatypes.data:ZarrDirectory">
         <infer_from suffix="zarr" />

lib/galaxy/datatypes/registry.py

@@ -77,6 +77,7 @@
         self.config = config
         self.edam = edam
         self.datatypes_by_extension: Dict[str, Data] = {}
+        self.datatypes_by_format = {}
         self.datatypes_by_suffix_inferences = {}
         self.mimetypes_by_extension = {}
         self.datatype_converters = {}
@@ -276,13 +277,25 @@
                             upload_warning_template = Template(upload_warning_el.text or "")
                         datatype_instance = datatype_class()
                         self.datatypes_by_extension[extension] = datatype_instance
+                        if not datatype_class.is_subclass:
+                            edam_format = datatype_class.edam_format
+                            prefixed_format = f"edam:{edam_format}"
+                            if prefixed_format not in self.datatypes_by_format:
+                                register_datatype_by_format = True
+                                for super_klass in datatype_class.__mro__[1:-1]:
+                                    super_edam_format = getattr(super_klass, "edam_format", None)
+                                    if super_edam_format == edam_format:
+                                        register_datatype_by_format = False
+                                        break
+                                if register_datatype_by_format:
+                                    self.datatypes_by_format[prefixed_format] = datatype_instance
                         if mimetype is None:
                             # Use default mimetype per datatype specification.
                             mimetype = self.datatypes_by_extension[extension].get_mime()
                         self.mimetypes_by_extension[extension] = mimetype
                         if datatype_class.track_type:
                             self.available_tracks.append(extension)
-                        if display_in_upload and extension not in self.upload_file_formats:
+                        if display_in_upload:
                             self.upload_file_formats.append(extension)
                         # Max file size cut off for setting optional metadata.
                         self.datatypes_by_extension[extension].max_optional_metadata_filesize = elem.get(
@@ -443,6 +456,7 @@
                 override=override,
                 compressed_sniffers=compressed_sniffers,
             )
+            self.upload_file_formats = list(set(self.upload_file_formats))
             self.upload_file_formats.sort()
             # Load build sites
             if use_build_sites:
@@ -615,7 +629,7 @@
 
         return generic_datatype_instance
 
     def is_extension_unsniffable_binary(self, ext):
         datatype = self.get_datatype_by_extension(ext)
         return datatype is not None and isinstance(datatype, binary.Binary) and not hasattr(datatype, "sniff")
 
@@ -654,6 +668,20 @@
         """Returns a datatype object based on an extension"""
         return self.datatypes_by_extension.get(ext, None)
 
+    def get_datatype_by_format_ontology(self, ontology: str):
+        """Returns a datatype by format ontology"""
+        if "edamontology.org/" in ontology:
+            ontology = f"edam:{ontology.split('edamontology.org/')[1]}"
+        return self.datatypes_by_format.get(ontology)
+
+    def get_datatype_ext_by_format_ontology(self, ontology: str, only_uploadable: bool = False) -> Optional[str]:
+        """Returns a datatype by format ontology"""
+        datatype = self.get_datatype_by_format_ontology(ontology)
+        if datatype:
+            if not only_uploadable or datatype.file_ext in self.upload_file_formats:
+                return datatype.file_ext
+        return None
+
     def change_datatype(self, data, ext):
         if data.extension != ext:
             data.extension = ext

lib/galaxy/jobs/__init__.py

@@ -1172,7 +1172,7 @@ def can_split(self):
 
     @property
     def is_cwl_job(self):
-        return self.tool.tool_type == "cwl"
+        return self.tool.tool_type in ["galactic_cwl", "cwl"]
 
     def get_job_runner_url(self):
         log.warning(f"({self.job_id}) Job runner URLs are deprecated, use destinations instead.")

lib/galaxy/jobs/command_factory.py

@@ -94,19 +94,26 @@ def build_command(
             external_command_shell = container.shell
         else:
             external_command_shell = shell
-        externalized_commands = __externalize_commands(
-            job_wrapper, external_command_shell, commands_builder, remote_command_params, container=container
-        )
         if container and modify_command_for_container:
-            # Stop now and build command before handling metadata and copying
-            # working directory files back. These should always happen outside
-            # of docker container - no security implications when generating
-            # metadata and means no need for Galaxy to be available to container
-            # and not copying workdir outputs back means on can be more restrictive
-            # of where container can write to in some circumstances.
-            run_in_container_command = container.containerize_command(externalized_commands)
+            if job_wrapper.tool and not job_wrapper.tool.may_use_container_entry_point:
+                externalized_commands = __externalize_commands(
+                    job_wrapper, external_command_shell, commands_builder, remote_command_params, container=container
+                )
+                # Stop now and build command before handling metadata and copying
+                # working directory files back. These should always happen outside
+                # of docker container - no security implications when generating
+                # metadata and means no need for Galaxy to be available to container
+                # and not copying workdir outputs back means on can be more restrictive
+                # of where container can write to in some circumstances.
+                run_in_container_command = container.containerize_command(externalized_commands)
+            else:
+                tool_commands = commands_builder.build()
+                run_in_container_command = container.containerize_command(tool_commands)
             commands_builder = CommandsBuilder(run_in_container_command)
         else:
+            externalized_commands = __externalize_commands(
+                job_wrapper, external_command_shell, commands_builder, remote_command_params, container=container
+            )
             commands_builder = CommandsBuilder(externalized_commands)
 
     # Galaxy writes I/O files to outputs, Pulsar uses metadata. metadata seems like

lib/galaxy/jobs/runners/local.py

@@ -4,6 +4,7 @@
 
 import datetime
 import logging
+import math
 import os
 import subprocess
 import tempfile
@@ -67,7 +68,16 @@ def _command_line(self, job_wrapper: "MinimalJobWrapper") -> Tuple[str, str]:
         if slots:
             slots_statement = f'GALAXY_SLOTS="{int(slots)}"; export GALAXY_SLOTS; GALAXY_SLOTS_CONFIGURED="1"; export GALAXY_SLOTS_CONFIGURED;'
         else:
-            slots_statement = 'GALAXY_SLOTS="1"; export GALAXY_SLOTS;'
+            cores_min = 1
+            if job_wrapper.tool:
+                try:
+                    # In CWL 1.2 it can be a float that can be rounded to the next whole number
+                    cores_min = math.ceil(float(job_wrapper.tool.cores_min))
+                except ValueError:
+                    # TODO: in CWL this can be an expression referencing runtime
+                    # parameters, e.g. `$(inputs.special_file.size)`
+                    pass
+            slots_statement = f'GALAXY_SLOTS="{cores_min}"; export GALAXY_SLOTS;'
 
         job_id = job_wrapper.get_id_tag()
         job_file = JobState.default_job_file(job_wrapper.working_directory, job_id)

lib/galaxy/managers/collections.py

@@ -258,8 +258,11 @@ def _create_instance_for_collection(
                 name=name,
             )
             assert isinstance(dataset_collection_instance, model.HistoryDatasetCollectionAssociation)
+
             if implicit_inputs:
                 for input_name, input_collection in implicit_inputs:
+                    if getattr(input_collection, "ephemeral", False):
+                        input_collection = input_collection.persistent_object
                     if isinstance(input_collection, model.HistoryDatasetCollectionAssociation):
                         # Can also get dragged DatasetCollectionElement's here.
                         # We only use this for extracting workflows currently,
@@ -424,6 +427,8 @@ def _append_tags(self, dataset_collection_instance, implicit_inputs=None, tags=N
         tags = tags or {}
         implicit_inputs = implicit_inputs or []
         for _, v in implicit_inputs:
+            if getattr(v, "ephemeral", False):
+                v = v.persistent_object
             for tag in v.auto_propagated_tags:
                 tags[tag.value] = tag
         for _, tag in tags.items():

lib/galaxy/managers/datasets.py

@@ -670,6 +670,7 @@ def add_serializers(self):
             "genome_build": lambda item, key, **context: str(item.dbkey) if item.dbkey is not None else None,
             # derived (not mapped) attributes
             "data_type": lambda item, key, **context: f"{item.datatype.__class__.__module__}.{item.datatype.__class__.__name__}",
+            "cwl_formats": lambda item, key, **context: item.cwl_formats,
             "converted": self.serialize_converted_datasets,
             # TODO: metadata/extra files
         }

lib/galaxy/managers/executables.py

@@ -29,22 +29,25 @@ def artifact_class(trans, as_dict: Dict[str, Any], allow_in_directory: Optional[
             as_dict = yaml.safe_load(f)
 
     artifact_class = as_dict.get("class", None)
+    target_object = None
     if artifact_class is None and "$graph" in as_dict:
         object_id = object_id or "main"
         graph = as_dict["$graph"]
-        target_object = None
         if isinstance(graph, dict):
             target_object = graph.get(object_id)
         else:
             for item in graph:
                 found_id = item.get("id")
                 if found_id == object_id or found_id == f"#{object_id}":
                     target_object = item
+                    break
 
         if target_object and target_object.get("class"):
             artifact_class = target_object["class"]
+            if artifact_class in ("CommandLineTool", "ExpressionTool"):
+                target_object["cwlVersion"] = as_dict["cwlVersion"]
 
-    return artifact_class, as_dict, object_id
+    return artifact_class, as_dict, object_id, target_object
 
 
 __all__ = ("artifact_class",)

lib/galaxy/managers/hdas.py

@@ -546,6 +546,7 @@ def __init__(self, app: StructuredApp):
                 "file_name",
                 "display_apps",
                 "display_types",
+                "cwl_formats",
                 "validated_state",
                 "validated_state_message",
                 # 'url',

lib/galaxy/managers/tools.py

@@ -45,6 +45,7 @@
 
 if TYPE_CHECKING:
     from galaxy.managers.base import OrmFilterParsersType
+    from galaxy.tool_util.cwl.parser import ToolProxy
 
 
 def tool_payload_to_tool(app, tool_dict: Dict[str, Any]) -> Optional[Tool]:
@@ -96,17 +97,18 @@ def get_tool_by_id(self, object_id):
         stmt = select(DynamicTool).where(DynamicTool.id == object_id, DynamicTool.public == true())
         return self.session().scalars(stmt).one_or_none()
 
-    def create_tool(self, tool_payload: DynamicToolPayload):
+    def create_tool(self, tool_payload: DynamicToolPayload) -> DynamicTool:
         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."
             )
 
         uuid = model.get_uuid()
+        target_object = None
         tool_directory: Optional[str] = None
         tool_path: Optional[str] = None
         if tool_payload.src == "from_path":
-            tool_format, representation, _ = artifact_class(None, tool_payload.model_dump())
+            tool_format, representation, _, target_object = artifact_class(None, tool_payload.model_dump())
             tool_directory = tool_payload.tool_directory
             tool_path = tool_payload.path
         else:
@@ -118,13 +120,17 @@ def create_tool(self, tool_payload: DynamicToolPayload):
             if not tool_format:
                 raise exceptions.ObjectAttributeMissingException("Current tool representations require 'class'.")
 
+        proxy: Optional[ToolProxy] = 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 tool_path:
+            if target_object is not None:
+                representation = {"raw_process_reference": target_object, "uuid": str(uuid), "class": tool_format}
+                proxy = tool_proxy(tool_object=target_object, tool_directory=tool_directory, uuid=uuid)
+            elif tool_path:
                 proxy = tool_proxy(tool_path=tool_path, uuid=uuid)
             else:
                 # Build a tool proxy so that we can convert to the persistable
@@ -149,10 +155,31 @@ def create_tool(self, tool_payload: DynamicToolPayload):
             hidden=tool_payload.hidden,
             value=representation,
             public=True,
+            proxy=proxy,
         )
         self.app.toolbox.load_dynamic_tool(dynamic_tool)
         return dynamic_tool
 
+    def create_tool_from_proxy(self, proxy: "ToolProxy") -> DynamicTool:
+        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."
+            )
+        dynamic_tool = self.get_tool_by_uuid(proxy.uuid)
+        if not dynamic_tool:
+            representation = proxy.to_persistent_representation()
+            dynamic_tool = self.create(
+                tool_format=proxy._class,
+                tool_id=proxy.galaxy_id(),
+                tool_version=representation.get("version"),
+                uuid=proxy.uuid,
+                value=representation,
+                public=True,
+                proxy=proxy,
+            )
+        self.app.toolbox.load_dynamic_tool(dynamic_tool)
+        return dynamic_tool
+
     def create_unprivileged_tool(
         self, user: model.User, tool_payload: DynamicUnprivilegedToolCreatePayload
     ) -> DynamicTool: