From 0f185683b7c574061b4e7fa65484989867b421ee Mon Sep 17 00:00:00 2001 From: "Michael R. Crusoe" Date: Mon, 8 Nov 2021 20:28:35 +0100 Subject: [PATCH] improve typing of lib/galaxy/workflow/modules.py --- lib/galaxy/workflow/modules.py | 88 +++++++++++++++++++++------------- setup.cfg | 2 - 2 files changed, 54 insertions(+), 36 deletions(-) diff --git a/lib/galaxy/workflow/modules.py b/lib/galaxy/workflow/modules.py index cfe291eeea2..f259f3a5c8e 100644 --- a/lib/galaxy/workflow/modules.py +++ b/lib/galaxy/workflow/modules.py @@ -5,6 +5,7 @@ import json import logging import re from collections import defaultdict +from typing import Any, cast, Dict, List, Optional, Union import packaging.version @@ -15,7 +16,7 @@ from galaxy import ( ) from galaxy.exceptions import ToolMissingException from galaxy.jobs.actions.post import ActionBox -from galaxy.model import PostJobAction +from galaxy.model import PostJobAction, Workflow from galaxy.model.dataset_collections import matching from galaxy.tool_util.parser.output_objects import ToolExpressionOutput from galaxy.tools import ( @@ -82,6 +83,10 @@ NO_REPLACEMENT = NoReplacement() class WorkflowModule: + label: str + type: str + name: str + def __init__(self, trans, content_id=None, **kwds): self.trans = trans self.content_id = content_id @@ -398,7 +403,8 @@ class SubWorkflowModule(WorkflowModule): # - Second pass actually turn RuntimeInputs into inputs if possible. type = "subworkflow" name = "Subworkflow" - _modules = None + _modules: Optional[List[Any]] = None + subworkflow: Workflow @classmethod def from_dict(Class, trans, d, **kwds): @@ -474,10 +480,12 @@ class SubWorkflowModule(WorkflowModule): return {f"Step {i + 1}": upgrade_message for i, upgrade_message in enumerate(states) if upgrade_message} or None def get_errors(self, **kwargs): - errors = (module.get_errors(include_tool_id=True) for module in self.get_modules()) - errors = [e for e in errors if e] - if any(errors): - return errors + errors1 = ( + module.get_errors(include_tool_id=True) for module in self.get_modules() + ) + errors2 = [e for e in errors1 if e] + if any(errors2): + return errors2 return None def get_all_outputs(self, data_only=False): @@ -830,42 +838,50 @@ class InputParameterModule(WorkflowModule): cases = [] for param_type in ["text", "integer", "float", "boolean", "color"]: - default_source = dict(name="default", label="Default Value", type=param_type) + default_source: Dict[str, Union[int, float, bool, str]] = dict( + name="default", label="Default Value", type=param_type + ) if param_type == "text": if parameter_type == "text": - default = parameter_def.get("default") or "" + text_default = parameter_def.get("default") or "" else: - default = "" - default_source["value"] = default - input_default_value = TextToolParameter(None, default_source) + text_default = "" + default_source["value"] = text_default + input_default_value: Union[ + TextToolParameter, + IntegerToolParameter, + FloatToolParameter, + BooleanToolParameter, + ColorToolParameter, + ] = TextToolParameter(None, default_source) elif param_type == "integer": if parameter_type == "integer": - default = parameter_def.get("default") or 0 + integer_default = parameter_def.get("default") or 0 else: - default = 0 - default_source["value"] = default + integer_default = 0 + default_source["value"] = integer_default input_default_value = IntegerToolParameter(None, default_source) elif param_type == "float": if parameter_type == "float": - default = parameter_def.get("default") or 0.0 + float_default = parameter_def.get("default") or 0.0 else: - default = 0.0 - default_source["value"] = default + float_default = 0.0 + default_source["value"] = float_default input_default_value = FloatToolParameter(None, default_source) elif param_type == "boolean": if parameter_type == "boolean": - default = parameter_def.get("default") or False + boolean_default = parameter_def.get("default") or False else: - default = False - default_source["value"] = default - default_source["checked"] = default + boolean_default = False + default_source["value"] = boolean_default + default_source["checked"] = boolean_default input_default_value = BooleanToolParameter(None, default_source) elif param_type == "color": - if parameter_type == 'color': - default = parameter_def.get('default') or '#000000' + if parameter_type == "color": + color_default = parameter_def.get("default") or "#000000" else: - default = '#000000' - default_source["value"] = default + color_default = "#000000" + default_source["value"] = color_default input_default_value = ColorToolParameter(None, default_source) optional_value = optional_param(optional) @@ -910,7 +926,9 @@ class InputParameterModule(WorkflowModule): optional_cond.cases = optional_cases if param_type == "text": - restrict_how_source = dict(name="how", label="Restrict Text Values?", type="select") + restrict_how_source: Union[ + Dict[str, Union[str, List[Dict[str, Union[str, bool]]]]] + ] = dict(name="how", label="Restrict Text Values?", type="select") if parameter_def.get("restrictions") is not None: restrict_how_value = "staticRestrictions" elif parameter_def.get("restrictOnConnections") is True: @@ -979,7 +997,7 @@ class InputParameterModule(WorkflowModule): raise ValueError("Invalid parameter type for workflow parameters encountered.") # Optional parameters for tool input source definition. - parameter_kwds = {} + parameter_kwds: Dict[str, Union[str, List[Dict[str, Any]]]] = {} is_text = parameter_type == "text" restricted_inputs = False @@ -1506,7 +1524,7 @@ class ToolModule(WorkflowModule): def get_config_form(self, step=None): if self.tool: self.add_dummy_datasets(connections=step and step.input_connections) - incoming = {} + incoming: Dict[str, str] = {} params_to_incoming(incoming, self.tool.inputs, self.state.inputs, self.trans.app) return self.tool.to_json(self.trans, incoming, workflow_building_mode=True) @@ -1698,8 +1716,8 @@ class ToolModule(WorkflowModule): def callback(input, prefixed_name, **kwargs): input_dict = all_inputs_by_name[prefixed_name] - replacement = NO_REPLACEMENT - dataset_instance = None + replacement: Union[model.Dataset, NoReplacement] = NO_REPLACEMENT + dataset_instance: Optional[model.Dataset] = None if iteration_elements and prefixed_name in iteration_elements: dataset_instance = getattr(iteration_elements[prefixed_name], 'dataset_instance', None) if isinstance(input, DataToolParameter) and dataset_instance: @@ -1707,7 +1725,7 @@ class ToolModule(WorkflowModule): # See https://github.com/galaxyproject/galaxy/pull/1693 for context. replacement = dataset_instance if hasattr(iteration_elements[prefixed_name], 'element_identifier') and iteration_elements[prefixed_name].element_identifier: - replacement.element_identifier = iteration_elements[prefixed_name].element_identifier + replacement.element_identifier = iteration_elements[prefixed_name].element_identifier # type: ignore else: # If collection - just use element model object. replacement = iteration_elements[prefixed_name] @@ -1717,9 +1735,11 @@ class ToolModule(WorkflowModule): if replacement is not NO_REPLACEMENT: if not isinstance(input, BaseDataToolParameter): # Probably a parameter that can be replaced - dataset = dataset_instance or replacement - if getattr(dataset, 'extension', None) == 'expression.json': - with open(dataset.file_name) as f: + dataset2: model.Dataset = cast( + model.Dataset, dataset_instance or replacement + ) + if getattr(dataset2, "extension", None) == "expression.json": + with open(dataset2.file_name) as f: replacement = json.load(f) found_replacement_keys.add(prefixed_name) diff --git a/setup.cfg b/setup.cfg index 5161894188d..53f7bc4b492 100644 --- a/setup.cfg +++ b/setup.cfg @@ -574,8 +574,6 @@ check_untyped_defs = False check_untyped_defs = False [mypy-galaxy.jobs.mapper] check_untyped_defs = False -[mypy-galaxy.workflow.modules] -check_untyped_defs = False [mypy-galaxy.tools.evaluation] check_untyped_defs = False [mypy-galaxy.jobs.runners]