cwltool.subgraph

Attributes

CWLObjectType

Typical raw dictionary found in lightly parsed CWL.

Node

UP

DOWN

INPUT

OUTPUT

STEP

Classes

LoadingContext

Shared kwargs based initializer for RuntimeContext and LoadingContext.

Process

Abstract CWL Process.

Workflow

Abstract CWL Process.

WorkflowStep

Abstract CWL Process.

Functions

load_tool(argsworkflow[, loadingContext])

make_tool(uri, loadingContext)

Make a Python CWL object.

aslist(thing)

Wrap any non-MutableSequence/list in a list.

subgraph_visit(current, nodes, visited, direction)

declare_node(nodes, nodeid, tp)

find_step(steps, stepid, loading_context)

Find the step (raw dictionary and WorkflowStep) for a given step id.

get_subgraph(roots, tool, loading_context)

Extract the subgraph for the given roots.

get_step(tool, step_id, loading_context)

Extract a single WorkflowStep for the given step_id.

get_process(tool, step_id, loading_context)

Find the underlying Process for a given Workflow step id.

Module Contents

class cwltool.subgraph.LoadingContext(kwargs=None)

Bases: ContextBase

Inheritance diagram of cwltool.subgraph.LoadingContext

Shared kwargs based initializer for RuntimeContext and LoadingContext.

Parameters:

kwargs (Optional[Dict[str, Any]])

copy()

Return a copy of this LoadingContext.

Return type:

LoadingContext

cwltool.subgraph.load_tool(argsworkflow, loadingContext=None)
Parameters:
Return type:

cwltool.process.Process

cwltool.subgraph.make_tool(uri, loadingContext)

Make a Python CWL object.

Parameters:
Return type:

cwltool.process.Process

class cwltool.subgraph.Process(toolpath_object, loadingContext)

Bases: cwltool.utils.HasReqsHints

Inheritance diagram of cwltool.subgraph.Process

Abstract CWL Process.

Parameters:
evalResources(builder, runtimeContext)
Parameters:
Return type:

Dict[str, Union[int, float]]

checkRequirements(rec, supported_process_requirements)

Check the presence of unsupported requirements.

Parameters:
  • rec (Union[MutableSequence[cwltool.utils.CWLObjectType], cwltool.utils.CWLObjectType, cwltool.utils.CWLOutputType, None])

  • supported_process_requirements (Iterable[str])

Return type:

None

validate_hints(avsc_names, hints, strict)

Process the hints field.

Parameters:
Return type:

None

visit(op)
Parameters:

op (Callable[[ruamel.yaml.comments.CommentedMap], None])

Return type:

None

abstract job(job_order, output_callbacks, runtimeContext)
Parameters:
  • job_order (cwltool.utils.CWLObjectType)

  • output_callbacks (Optional[cwltool.utils.OutputCallbackType])

  • runtimeContext (cwltool.context.RuntimeContext)

Return type:

cwltool.utils.JobsGeneratorType

__str__()

Return the id of this CWL process.

Return type:

str

cwltool.subgraph.CWLObjectType

Typical raw dictionary found in lightly parsed CWL.

cwltool.subgraph.aslist(thing)

Wrap any non-MutableSequence/list in a list.

Parameters:

thing (Any)

Return type:

MutableSequence[Any]

class cwltool.subgraph.Workflow(toolpath_object, loadingContext)

Bases: cwltool.process.Process

Inheritance diagram of cwltool.subgraph.Workflow

Abstract CWL Process.

Parameters:
make_workflow_step(toolpath_object, pos, loadingContext, parentworkflowProv=None)
Parameters:
Return type:

WorkflowStep

job(job_order, output_callbacks, runtimeContext)
Parameters:
  • job_order (cwltool.utils.CWLObjectType)

  • output_callbacks (Optional[cwltool.utils.OutputCallbackType])

  • runtimeContext (cwltool.context.RuntimeContext)

Return type:

cwltool.utils.JobsGeneratorType

visit(op)
Parameters:

op (Callable[[ruamel.yaml.comments.CommentedMap], None])

Return type:

None

class cwltool.subgraph.WorkflowStep(toolpath_object, pos, loadingContext, parentworkflowProv=None)

Bases: cwltool.process.Process

Inheritance diagram of cwltool.subgraph.WorkflowStep

Abstract CWL Process.

Parameters:
checkRequirements(rec, supported_process_requirements)

Check the presence of unsupported requirements.

Parameters:
  • rec (Union[MutableSequence[cwltool.utils.CWLObjectType], cwltool.utils.CWLObjectType, cwltool.utils.CWLOutputType, None])

  • supported_process_requirements (Iterable[str])

Return type:

None

receive_output(output_callback, jobout, processStatus)
Parameters:
  • output_callback (cwltool.utils.OutputCallbackType)

  • jobout (cwltool.utils.CWLObjectType)

  • processStatus (str)

Return type:

None

job(job_order, output_callbacks, runtimeContext)

Initialize sub-workflow as a step in the parent profile.

Parameters:
  • job_order (cwltool.utils.CWLObjectType)

  • output_callbacks (Optional[cwltool.utils.OutputCallbackType])

  • runtimeContext (cwltool.context.RuntimeContext)

Return type:

cwltool.utils.JobsGeneratorType

visit(op)
Parameters:

op (Callable[[ruamel.yaml.comments.CommentedMap], None])

Return type:

None

cwltool.subgraph.Node
cwltool.subgraph.UP = 'up'
cwltool.subgraph.DOWN = 'down'
cwltool.subgraph.INPUT = 'input'
cwltool.subgraph.OUTPUT = 'output'
cwltool.subgraph.STEP = 'step'
cwltool.subgraph.subgraph_visit(current, nodes, visited, direction)
Parameters:
  • current (str)

  • nodes (MutableMapping[str, Node])

  • visited (Set[str])

  • direction (str)

Return type:

None

cwltool.subgraph.declare_node(nodes, nodeid, tp)
Parameters:
  • nodes (Dict[str, Node])

  • nodeid (str)

  • tp (Optional[str])

Return type:

Node

cwltool.subgraph.find_step(steps, stepid, loading_context)

Find the step (raw dictionary and WorkflowStep) for a given step id.

Parameters:
Return type:

Tuple[Optional[cwltool.utils.CWLObjectType], Optional[cwltool.workflow.WorkflowStep]]

cwltool.subgraph.get_subgraph(roots, tool, loading_context)

Extract the subgraph for the given roots.

Parameters:
Return type:

ruamel.yaml.comments.CommentedMap

cwltool.subgraph.get_step(tool, step_id, loading_context)

Extract a single WorkflowStep for the given step_id.

Parameters:
Return type:

ruamel.yaml.comments.CommentedMap

cwltool.subgraph.get_process(tool, step_id, loading_context)

Find the underlying Process for a given Workflow step id.

Parameters:
Return type:

Tuple[Any, cwltool.workflow.WorkflowStep]