cwltool.checker

Static checking of CWL workflow connectivity.

Module Contents

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 = False)

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

missing_subset(fullset, subset)

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.

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.

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)

Attributes

SrcSink

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

Check if the source and sink types are correct.

Acceptable types are “pass”, “warning”, or “exception”.

Parameters

  • srctype (cwltool.utils.SinkType) –

  • sinktype (cwltool.utils.SinkType) –

  • linkMerge (Optional[str]) –

  • valueFrom (Optional[str]) –

Return type

str

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.

src: admissible source types sink: admissible sink types

In non-strict comparison, at least one source type must match one sink type. In strict comparison, all source types must match at least one sink type.

Parameters

  • src (cwltool.utils.SinkType) –

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

  • strict (bool) –

Return type

bool

cwltool.checker.missing_subset(fullset, 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.

Parameters

  • workflow_inputs (List[cwltool.utils.CWLObjectType]) –

  • workflow_outputs (MutableSequence[cwltool.utils.CWLObjectType]) –

  • step_inputs (MutableSequence[cwltool.utils.CWLObjectType]) –

  • step_outputs (List[cwltool.utils.CWLObjectType]) –

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

Return type

None

cwltool.checker.SrcSink
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.

sourceField is either “source” or “outputSource”

Parameters

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

  • sinks (MutableSequence[cwltool.utils.CWLObjectType]) –

  • sourceField (str) –

  • 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.

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)
Parameters

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

  • parm_id (str) –

Return type

bool