cwltool.checker
Static checking of CWL workflow connectivity.
Attributes
Functions
|
Check if the source and sink types are correct. |
|
Return the merge flattened type of the source type. |
|
Check for identical type specifications, ignoring extra keys like inputBinding. |
|
|
|
Check if all source and sink types of a workflow are compatible before run time. |
|
Given a list of sinks, check if their types match with the types of their sources. |
|
Check if a workflow has circular dependency. |
|
Get the dependency tree in the form of adjacency list. |
|
Perform depth first search. |
|
Extract step id from either input or output fields. |
|
|
|
Check if a step contains a loop directive with all outputMethod. |
|
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)
- 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 (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.
- 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 (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)
- 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 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 (Iterator[MutableMapping[str, Any]])
- Return type:
None