opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-24 16:30:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

improve typing of lib/galaxy/workflow/modules.py

Browse Source
This commit is contained in:
Michael R. Crusoe
2021-11-11 11:55:05 +01:00
parent 03e796a94c
commit 0f185683b7
2 changed files with 54 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/workflow/modules.py
+54 -34
View File
@@ -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)
setup.cfg
-2
View File
@@ -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]