cwltool.checker

Static checking of CWL workflow connectivity.

Classes

SrcSink

Functions

check_types(srctype, sinktype, linkMerge, valueFrom)	Check if the source and sink types are correct.
merge_flatten_type(src)	Return the merge flattened type of the source type.
can_assign_src_to_sink(src, sink[, strict])	Check for identical type specifications, ignoring extra keys like inputBinding.
missing_subset(fullset, subset)	Calculate the items missing from the fullset given the subset.
static_checker(workflow_inputs, workflow_outputs, ...)	Check if all source and sink types of a workflow are compatible before run time.
check_all_types(src_dict, sinks, sourceField, ...)	Given a list of sinks, check if their types match with the types of their sources.
circular_dependency_checker(step_inputs)	Check if a workflow has circular dependency.
get_dependency_tree(step_inputs)	Get the dependency tree in the form of adjacency list.
processDFS(adjacency, traversal_path, processed, cycles)	Perform depth first search.
get_step_id(field_id)	Extract step id from either input or output fields.
is_conditional_step(param_to_step, parm_id)	Return True if the step given by the parm_id is a conditional step.
is_all_output_method_loop_step(param_to_step, parm_id)	Check if a step contains a loop directive with all_iterations outputMethod.
loop_checker(steps)	Check loop compatibility with other directives.

Module Contents

cwltool.checker.check_types(srctype, sinktype, linkMerge, valueFrom)

Check if the source and sink types are correct.

Raises:

WorkflowException – If there is an unrecognized linkMerge type

Parameters:

• srctype (cwltool.utils.SinkType)

• sinktype (cwltool.utils.SinkType)

• linkMerge (Optional[str])

• valueFrom (Optional[str])

Return type:

Union[Literal[‘pass’], Literal[‘warning’], Literal[‘exception’]]

cwltool.checker.merge_flatten_type(src)

Return the merge flattened type of the source type.

Parameters:

src (cwltool.utils.SinkType)

Return type:

cwltool.utils.CWLOutputType

cwltool.checker.can_assign_src_to_sink(src, sink, strict=False)

Check for identical type specifications, ignoring extra keys like inputBinding.

In non-strict comparison, at least one source type must match one sink type, except for ‘null’. In strict comparison, all source types must match at least one sink type.

Parameters:

• src (cwltool.utils.SinkType) – admissible source types

• sink (Optional[cwltool.utils.SinkType]) – admissible sink types

• strict (bool)

Return type:

bool

cwltool.checker.missing_subset(fullset, subset)

Calculate the items missing from the fullset given the subset.

Parameters:

• fullset (list[Any])

• subset (list[Any])

Return type:

list[Any]

cwltool.checker.static_checker(workflow_inputs, workflow_outputs, step_inputs, step_outputs, param_to_step)

Check if all source and sink types of a workflow are compatible before run time.

Raises:

ValidationException – If any incompatibilities are detected.

Parameters:

• workflow_inputs (list[cwltool.utils.CWLObjectType])

• workflow_outputs (collections.abc.MutableSequence[cwltool.utils.CWLObjectType])

• step_inputs (collections.abc.MutableSequence[cwltool.utils.CWLObjectType])

• step_outputs (list[cwltool.utils.CWLObjectType])

• param_to_step (dict[str, cwltool.utils.CWLObjectType])

Return type:

None

class cwltool.checker.SrcSink

Bases: tuple

src

sink

linkMerge

message

cwltool.checker.check_all_types(src_dict, sinks, sourceField, param_to_step)

Given a list of sinks, check if their types match with the types of their sources.

Raises:

• WorkflowException – if there is an unrecognized linkMerge value (from check_types())

• ValidationException – if a sourceField is missing

Parameters:

• src_dict (dict[str, cwltool.utils.CWLObjectType])

• sinks (collections.abc.MutableSequence[cwltool.utils.CWLObjectType])

• sourceField (Union[Literal['source'], Literal['outputSource']])

• param_to_step (dict[str, cwltool.utils.CWLObjectType])

Return type:

dict[str, list[SrcSink]]

cwltool.checker.circular_dependency_checker(step_inputs)

Check if a workflow has circular dependency.

Raises:

ValidationException – If a circular dependency is detected.

Parameters:

step_inputs (list[cwltool.utils.CWLObjectType])

Return type:

None

cwltool.checker.get_dependency_tree(step_inputs)

Get the dependency tree in the form of adjacency list.

Parameters:

step_inputs (list[cwltool.utils.CWLObjectType])

Return type:

dict[str, list[str]]

cwltool.checker.processDFS(adjacency, traversal_path, processed, cycles)

Perform depth first search.

Parameters:

• adjacency (dict[str, list[str]])

• traversal_path (list[str])

• processed (list[str])

• cycles (list[list[str]])

Return type:

None

cwltool.checker.get_step_id(field_id)

Extract step id from either input or output fields.

Parameters:

field_id (str)

Return type:

str

cwltool.checker.is_conditional_step(param_to_step, parm_id)

Return True if the step given by the parm_id is a conditional step.

Parameters:

• param_to_step (dict[str, cwltool.utils.CWLObjectType])

• parm_id (str)

Return type:

bool

cwltool.checker.is_all_output_method_loop_step(param_to_step, parm_id)

Check if a step contains a loop directive with all_iterations outputMethod.

Parameters:

• param_to_step (dict[str, cwltool.utils.CWLObjectType])

• parm_id (str)

Return type:

bool

cwltool.checker.loop_checker(steps)

Check loop compatibility with other directives.

Raises:

ValidationException – If there is an incompatible combination between loop and scatter.

Parameters:

steps (collections.abc.Iterator[collections.abc.MutableMapping[str, Any]])

Return type:

None