Workflow conversion and validation things.

Author: jmchilton

lib/galaxy/tool_util/parameters/__init__.py

@@ -135,6 +135,7 @@
     "RepeatParameterModel",
     "RawStateDict",
     "ValidationFunctionT",
+    "is_optional",
     "validate_against_model",
     "validate_internal_job",
     "validate_internal_landing_request",

lib/galaxy/tool_util/version_util.py

@@ -0,0 +1,10 @@
+from typing import Union
+
+from packaging.version import Version
+
+from .version import LegacyVersion
+
+AnyVersionT = Union[LegacyVersion, Version]
+
+
+__all__ = ["AnyVersionT"]

lib/galaxy/tool_util/workflow_state/__init__.py

@@ -0,0 +1,23 @@
+"""Abstractions for reasoning about tool state within Galaxy workflows.
+
+Like everything else in galaxy-tool-util, this package should be independent of
+Galaxy's runtime. It is meant to provide utilities for reasonsing about tool state
+(largely building on the abstractions in galaxy.tool_util.parameters) within the
+context of workflows.
+"""
+
+from ._types import GetToolInfo
+from .convert import (
+    ConversionValidationFailure,
+    convert_state_to_format2,
+    Format2State,
+)
+from .validation import validate_workflow
+
+__all__ = (
+    "ConversionValidationFailure",
+    "convert_state_to_format2",
+    "GetToolInfo",
+    "Format2State",
+    "validate_workflow",
+)

lib/galaxy/tool_util/workflow_state/_types.py

@@ -0,0 +1,28 @@
+from typing import (
+    Any,
+    Dict,
+    Optional,
+    Union,
+)
+
+from typing_extensions import (
+    Literal,
+    Protocol,
+)
+
+from galaxy.tool_util_models import ParsedTool
+
+NativeWorkflowDict = Dict[str, Any]
+Format2WorkflowDict = Dict[str, Any]
+AnyWorkflowDict = Union[NativeWorkflowDict, Format2WorkflowDict]
+WorkflowFormat = Literal["gxformat2", "native"]
+NativeStepDict = Dict[str, Any]
+Format2StepDict = Dict[str, Any]
+NativeToolStateDict = Dict[str, Any]
+Format2StateDict = Dict[str, Any]
+
+
+class GetToolInfo(Protocol):
+    """An interface for fetching tool information for steps in a workflow."""
+
+    def get_tool_info(self, tool_id: str, tool_version: Optional[str]) -> ParsedTool: ...

lib/galaxy/tool_util/workflow_state/convert.py

@@ -0,0 +1,134 @@
+from typing import (
+    Dict,
+    List,
+    Optional,
+)
+
+from pydantic import (
+    BaseModel,
+    Field,
+)
+
+from galaxy.tool_util.parameters import ToolParameterT
+from galaxy.tool_util_models import ParsedTool
+from ._types import (
+    Format2StateDict,
+    GetToolInfo,
+    NativeStepDict,
+)
+from .validation_format2 import validate_step_against
+from .validation_native import (
+    get_parsed_tool_for_native_step,
+    native_tool_state,
+    validate_native_step_against,
+)
+
+Format2InputsDictT = Dict[str, str]
+
+
+class Format2State(BaseModel):
+    state: Format2StateDict
+    inputs: Format2InputsDictT = Field(alias="in")
+
+
+class ConversionValidationFailure(Exception):
+    pass
+
+
+def convert_state_to_format2(native_step_dict: NativeStepDict, get_tool_info: GetToolInfo) -> Format2State:
+    parsed_tool = get_parsed_tool_for_native_step(native_step_dict, get_tool_info)
+    return convert_state_to_format2_using(native_step_dict, parsed_tool)
+
+
+def convert_state_to_format2_using(native_step_dict: NativeStepDict, parsed_tool: Optional[ParsedTool]) -> Format2State:
+    """Create a "clean" gxformat2 workflow tool state from a native workflow step.
+
+    gxformat2 does not know about tool specifications so it cannot reason about the native
+    tool state attribute and just copies it as is. This native state can be pretty ugly. The purpose
+    of this function is to build a cleaned up state to replace the gxformat2 copied native tool_state
+    with that is more readable and has stronger typing by using the tool's inputs to guide
+    the conversion (the parsed_tool parameter).
+
+    This method validates both the native tool state and the resulting gxformat2 tool state
+    so that we can be more confident the conversion doesn't corrupt the workflow. If no meta
+    model to validate against is supplied or if either validation fails this method throws
+    ConversionValidationFailure to signal the caller to just use the native tool state as is
+    instead of trying to convert it to a cleaner gxformat2 tool state - under the assumption
+    it is better to have an "ugly" workflow than a corrupted one during conversion.
+    """
+    if parsed_tool is None:
+        raise ConversionValidationFailure("Could not resolve tool inputs")
+    try:
+        validate_native_step_against(native_step_dict, parsed_tool)
+    except Exception:
+        raise ConversionValidationFailure(
+            "Failed to validate native step - not going to convert a tool state that isn't understood"
+        )
+    result = _convert_valid_state_to_format2(native_step_dict, parsed_tool)
+    print(result.dict())
+    try:
+        validate_step_against(result.dict(), parsed_tool)
+    except Exception:
+        raise ConversionValidationFailure(
+            "Failed to validate resulting cleaned step - not going to convert to an unvalidated tool state"
+        )
+    return result
+
+
+def _convert_valid_state_to_format2(native_step_dict: NativeStepDict, parsed_tool: ParsedTool) -> Format2State:
+    format2_state: Format2StateDict = {}
+    format2_in: Format2InputsDictT = {}
+
+    root_tool_state = native_tool_state(native_step_dict)
+    tool_inputs = parsed_tool.inputs
+    _convert_state_level(native_step_dict, tool_inputs, root_tool_state, format2_state, format2_in)
+    return Format2State(
+        **{
+            "state": format2_state,
+            "in": format2_in,
+        }
+    )
+
+
+def _convert_state_level(
+    step: NativeStepDict,
+    tool_inputs: List[ToolParameterT],
+    native_state: dict,
+    format2_state_at_level: dict,
+    format2_in: Format2InputsDictT,
+    prefix: Optional[str] = None,
+) -> None:
+    prefix = prefix or ""
+    assert prefix is not None
+    for tool_input in tool_inputs:
+        _convert_state_at_level(step, tool_input, native_state, format2_state_at_level, format2_in, prefix)
+
+
+def _convert_state_at_level(
+    step: NativeStepDict,
+    tool_input: ToolParameterT,
+    native_state_at_level: dict,
+    format2_state_at_level: dict,
+    format2_in: Format2InputsDictT,
+    prefix: str,
+) -> None:
+    parameter_type = tool_input.parameter_type
+    parameter_name = tool_input.name
+    value = native_state_at_level.get(parameter_name, None)
+    state_path = parameter_name if prefix is None else f"{prefix}|{parameter_name}"
+    if parameter_type == "gx_integer":
+        # check for runtime input
+        try:
+            format2_value = int(value)  # type: ignore[arg-type]
+        except ValueError:
+            raise Exception(f"Failed to convert integer value {value} for parameter {parameter_name}")
+        format2_state_at_level[parameter_name] = format2_value
+    elif parameter_type == "gx_data":
+        input_connections = step.get("input_connections", {})
+        print(state_path)
+        print(input_connections)
+        if state_path in input_connections:
+            format2_in[state_path] = "placeholder"
+    else:
+        pass
+        # raise NotImplementedError(f"Unhandled parameter type {parameter_type}")

lib/galaxy/tool_util/workflow_state/validation.py

@@ -0,0 +1,21 @@
+from ._types import (
+    AnyWorkflowDict,
+    GetToolInfo,
+    WorkflowFormat,
+)
+from .validation_format2 import validate_workflow_format2
+from .validation_native import validate_workflow_native
+
+
+def validate_workflow(workflow_dict: AnyWorkflowDict, get_tool_info: GetToolInfo):
+    if _format(workflow_dict) == "gxformat2":
+        validate_workflow_format2(workflow_dict, get_tool_info)
+    else:
+        validate_workflow_native(workflow_dict, get_tool_info)
+
+
+def _format(workflow_dict: AnyWorkflowDict) -> WorkflowFormat:
+    if workflow_dict.get("a_galaxy_workflow") == "true":
+        return "native"
+    else:
+        return "gxformat2"

lib/galaxy/tool_util/workflow_state/validation_format2.py

@@ -0,0 +1,142 @@
+from typing import (
+    cast,
+    Optional,
+)
+
+from gxformat2.model import (
+    get_native_step_type,
+    pop_connect_from_step_dict,
+    setup_connected_values,
+    steps_as_list,
+)
+
+from galaxy.tool_util.parameters import (
+    ConditionalParameterModel,
+    ConditionalWhen,
+    flat_state_path,
+    keys_starting_with,
+    repeat_inputs_to_array,
+    RepeatParameterModel,
+    ToolParameterT,
+    validate_explicit_conditional_test_value,
+    WorkflowStepLinkedToolState,
+    WorkflowStepToolState,
+)
+from galaxy.tool_util_models import ParsedTool
+from ._types import (
+    Format2StepDict,
+    Format2WorkflowDict,
+    GetToolInfo,
+)
+
+
+def validate_workflow_format2(workflow_dict: Format2WorkflowDict, get_tool_info: GetToolInfo):
+    steps = steps_as_list(workflow_dict)
+    for step in steps:
+        validate_step_format2(step, get_tool_info)
+
+
+def validate_step_format2(step_dict: Format2StepDict, get_tool_info: GetToolInfo):
+    step_type = get_native_step_type(step_dict)
+    if step_type != "tool":
+        return
+    tool_id = cast(str, step_dict.get("tool_id"))
+    tool_version: Optional[str] = cast(Optional[str], step_dict.get("tool_version"))
+    parsed_tool = get_tool_info.get_tool_info(tool_id, tool_version)
+    if parsed_tool is not None:
+        validate_step_against(step_dict, parsed_tool)
+
+
+def validate_step_against(step_dict: Format2StepDict, parsed_tool: ParsedTool):
+    source_tool_state_model = WorkflowStepToolState.parameter_model_for(parsed_tool.inputs)
+    linked_tool_state_model = WorkflowStepLinkedToolState.parameter_model_for(parsed_tool.inputs)
+    contains_format2_state = "state" in step_dict
+    contains_native_state = "tool_state" in step_dict
+    if contains_format2_state:
+        assert source_tool_state_model
+        source_tool_state_model.model_validate(step_dict["state"])
+    if not contains_native_state:
+        if not contains_format2_state:
+            step_dict["state"] = {}
+        # setup links and then validate against model...
+        linked_step = merge_inputs(step_dict, parsed_tool)
+        linked_tool_state_model.model_validate(linked_step["state"])
+
+
+def merge_inputs(step_dict: Format2StepDict, parsed_tool: ParsedTool) -> Format2StepDict:
+    connect = pop_connect_from_step_dict(step_dict)
+    step_dict = setup_connected_values(step_dict, connect)
+    tool_inputs = parsed_tool.inputs
+
+    state_at_level = step_dict["state"]
+
+    for tool_input in tool_inputs:
+        _merge_into_state(connect, tool_input, state_at_level)
+
+    for key in connect:
+        raise Exception(f"Failed to find parameter definition matching workflow linked key {key}")
+    return step_dict
+
+
+def _merge_into_state(
+    connect, tool_input: ToolParameterT, state: dict, prefix: Optional[str] = None, branch_connect=None
+):
+    if branch_connect is None:
+        branch_connect = connect
+
+    name = tool_input.name
+    parameter_type = tool_input.parameter_type
+    state_path = flat_state_path(name, prefix)
+    if parameter_type == "gx_conditional":
+        conditional_state = state.get(name, {})
+        if name not in state:
+            state[name] = conditional_state
+
+        conditional = cast(ConditionalParameterModel, tool_input)
+        when: ConditionalWhen = _select_which_when(conditional, conditional_state)
+        test_parameter = conditional.test_parameter
+        conditional_connect = keys_starting_with(branch_connect, state_path)
+        _merge_into_state(
+            connect, test_parameter, conditional_state, prefix=state_path, branch_connect=conditional_connect
+        )
+        for when_parameter in when.parameters:
+            _merge_into_state(
+                connect, when_parameter, conditional_state, prefix=state_path, branch_connect=conditional_connect
+            )
+    elif parameter_type == "gx_repeat":
+        repeat_state_array = state.get(name, [])
+        repeat = cast(RepeatParameterModel, tool_input)
+        repeat_instance_connects = repeat_inputs_to_array(state_path, connect)
+        for i, repeat_instance_connect in enumerate(repeat_instance_connects):
+            while len(repeat_state_array) <= i:
+                repeat_state_array.append({})
+
+            repeat_instance_prefix = f"{state_path}_{i}"
+            for repeat_parameter in repeat.parameters:
+                _merge_into_state(
+                    connect,
+                    repeat_parameter,
+                    repeat_state_array[i],
+                    prefix=repeat_instance_prefix,
+                    branch_connect=repeat_instance_connect,
+                )
+        if repeat_state_array and name not in state:
+            state[name] = repeat_state_array
+    else:
+        if state_path in branch_connect:
+            state[name] = {"__class__": "ConnectedValue"}
+            del connect[state_path]
+
+
+def _select_which_when(conditional: ConditionalParameterModel, state: dict) -> ConditionalWhen:
+    test_parameter = conditional.test_parameter
+    test_parameter_name = test_parameter.name
+    explicit_test_value = state.get(test_parameter_name)
+    test_value = validate_explicit_conditional_test_value(test_parameter_name, explicit_test_value)
+    for when in conditional.whens:
+        if test_value is None and when.is_default_when:
+            return when
+        elif test_value == when.discriminator:
+            return when
+    else:
+        raise Exception(f"Invalid conditional test value ({explicit_test_value}) for parameter ({test_parameter_name})")