Module dewret.renderers.cwl

CWL Renderer.

Outputs a Common Workflow Language representation of the current workflow.

Variables

InputSchemaType

Functions

raw_to_command_input_schema

def raw_to_command_input_schema(
    label: str,
    value: Union[str, float, bool, bytes, int, NoneType, list['RawType'], dict[str, 'RawType']]
) -> Union[str, ForwardRef('CommandInputSchema'), list[str], list['InputSchemaType']]

Infer the CWL input structure for this value.

Inspects the value, to work out an appropriate structure describing it in CWL.

Parameters:

Name	Type	Description	Default
label	None	name of the variable.	None
value	None	basic-typed variable from which to build structure.	None

Returns:

Type	Description
None	Structure used to define (possibly compound) basic types for input.

render

def render(
    workflow: dewret.workflow.Workflow
) -> dict[str, typing.Union[str, float, bool, bytes, int, NoneType, list['RawType'], dict[str, 'RawType']]]

Render to a dict-like structure.

Parameters:

Name	Type	Description	Default
workflow	None	workflow to evaluate result.	None

Returns:

Type	Description
None	Reduced form as a native Python dict structure for serialization.

to_cwl_type

def to_cwl_type(
    typ: type
) -> str | list[str]

Map Python types to CWL types.

Parameters:

Name	Type	Description	Default
typ	None	a Python basic type.	None

Returns:

Type	Description
None	CWL specification type name, or a list if a union.

to_output_schema

def to_output_schema(
    label: str,
    typ: type[typing.Union[str, float, bool, bytes, int, NoneType, list['RawType'], dict[str, 'RawType'], attr.AttrsInstance, dewret.utils.DataclassProtocol]],
    output_source: str | None = None
) -> dewret.renderers.cwl.CommandOutputSchema

Turn a step's output into an output schema.

Takes a source, type and label and provides a description for CWL.

Parameters:

Name	Type	Description	Default
label	None	name of this field.	None
typ	None	either a basic type, compound of basic types, or a TypedDict representing a pre-defined result structure.	None
output_source	None	if provided, a CWL step result reference to input here.	None

Returns:

Type	Description
None	CWL CommandOutputSchema-like structure for embedding into an outputs block

Classes

CommandInputSchema

class CommandInputSchema(
    /,
    *args,
    **kwargs
)

Structure for referring to a raw type in CWL.

Encompasses several CWL types. In future, it may be best to use cwltool or another library for these basic structures.

Attributes

Name	Type	Description	Default
type	None	CWL type of this input.	None
label	None	name to show for this input.	None
fields	None	(for record) individual fields in a dict-like structure.	None
items	None	(for array) type that each field will have.	None

Ancestors (in MRO)

• builtins.dict

Methods

clear

def clear(
    ...
)

D.clear() -> None. Remove all items from D.

copy

def copy(
    ...
)

D.copy() -> a shallow copy of D

fromkeys

def fromkeys(
    iterable,
    value=None,
    /
)

Create a new dictionary with keys from iterable and values set to value.

get

def get(
    self,
    key,
    default=None,
    /
)

Return the value for key if key is in the dictionary, else default.

items

def items(
    ...
)

D.items() -> a set-like object providing a view on D's items

keys

def keys(
    ...
)

D.keys() -> a set-like object providing a view on D's keys

pop

def pop(
    ...
)

D.pop(k[,d]) -> v, remove specified key and return the corresponding value.

If the key is not found, return the default if given; otherwise, raise a KeyError.

popitem

def popitem(
    self,
    /
)

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

setdefault

def setdefault(
    self,
    key,
    default=None,
    /
)

Insert key with a value of default if key is not in the dictionary.

Return the value for key if key is in the dictionary, else default.

update

def update(
    ...
)

D.update([E, ]**F) -> None. Update D from dict/iterable E and F.

If E is present and has a .keys() method, then does: for k in E: D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

values

def values(
    ...
)

D.values() -> an object providing a view on D's values

CommandOutputSchema

class CommandOutputSchema(
    /,
    *args,
    **kwargs
)

Structure for referring to an output in CWL.

As a simplification, this is an input schema with an extra outputSource field.

Attributes

Name	Type	Description	Default
outputSource	None	step result to use for this output.	None

Ancestors (in MRO)

• builtins.dict

Methods

clear

def clear(
    ...
)

D.clear() -> None. Remove all items from D.

copy

def copy(
    ...
)

D.copy() -> a shallow copy of D

fromkeys

def fromkeys(
    iterable,
    value=None,
    /
)

Create a new dictionary with keys from iterable and values set to value.

get

def get(
    self,
    key,
    default=None,
    /
)

Return the value for key if key is in the dictionary, else default.

items

def items(
    ...
)

D.items() -> a set-like object providing a view on D's items

keys

def keys(
    ...
)

D.keys() -> a set-like object providing a view on D's keys

pop

def pop(
    ...
)

D.pop(k[,d]) -> v, remove specified key and return the corresponding value.

If the key is not found, return the default if given; otherwise, raise a KeyError.

popitem

def popitem(
    self,
    /
)

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

setdefault

def setdefault(
    self,
    key,
    default=None,
    /
)

Insert key with a value of default if key is not in the dictionary.

Return the value for key if key is in the dictionary, else default.

update

def update(
    ...
)

D.update([E, ]**F) -> None. Update D from dict/iterable E and F.

If E is present and has a .keys() method, then does: for k in E: D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

values

def values(
    ...
)

D.values() -> an object providing a view on D's values

InputsDefinition

class InputsDefinition(
    inputs: dict[str, 'CommandInputParameter']
)

CWL-renderable representation of an input parameter block.

Turns dewret results into a CWL input block.

Attributes

Name	Type	Description	Default
input	None	sequence of results from a workflow.	None

Class variables

CommandInputParameter

Static methods

from_parameters

def from_parameters(
    parameters: list[dewret.workflow.Parameter[typing.Union[str, float, bool, bytes, int, NoneType, list['RawType'], dict[str, 'RawType']]]]
) -> 'InputsDefinition'

Takes a list of parameters into a CWL structure.

Uses the parameters to fill out the necessary input fields.

Returns:

Type	Description
None	CWL-like structure representing all workflow outputs.

Instance variables

inputs

Methods

render

def render(
    self
) -> dict[str, typing.Union[str, float, bool, bytes, int, NoneType, list['RawType'], dict[str, 'RawType']]]

Render to a dict-like structure.

Returns:

Type	Description
None	Reduced form as a native Python dict structure for serialization.

OutputsDefinition

class OutputsDefinition(
    outputs: dict[str, 'CommandOutputSchema']
)

CWL-renderable set of workflow outputs.

Turns dewret results into a CWL output block.

Attributes

Name	Type	Description	Default
outputs	None	sequence of results from a workflow.	None

Static methods

from_results

def from_results(
    results: dict[str, dewret.workflow.StepReference[typing.Any]]
) -> 'OutputsDefinition'

Takes a mapping of results into a CWL structure.

Pulls the result type from the signature, ultimately, if possible.

Returns:

Type	Description
None	CWL-like structure representing all workflow outputs.

Instance variables

outputs

Methods

render

def render(
    self
) -> dict[str, typing.Union[str, float, bool, bytes, int, NoneType, list['RawType'], dict[str, 'RawType']]]

Render to a dict-like structure.

Returns:

Type	Description
None	Reduced form as a native Python dict structure for serialization.

ReferenceDefinition

class ReferenceDefinition(
    source: str
)

CWL-renderable internal reference.

Normally points to a value or a step.

Static methods

from_reference

def from_reference(
    ref: dewret.workflow.Reference
) -> 'ReferenceDefinition'

Build from a Reference.

Converts a dewret.workflow.Reference into a CWL-rendering object.

Parameters:

Name	Type	Description	Default
ref	None	reference to convert.	None

Instance variables

source

Methods

render

def render(
    self
) -> dict[str, typing.Union[str, float, bool, bytes, int, NoneType, list['RawType'], dict[str, 'RawType']]]

Render to a dict-like structure.

Returns:

Type	Description
None	Reduced form as a native Python dict structure for serialization.

StepDefinition

class StepDefinition(
    name: str,
    run: str,
    out: dict[str, 'CommandInputSchema'] | list[str],
    in_: collections.abc.Mapping[str, dewret.renderers.cwl.ReferenceDefinition | dewret.workflow.Raw]
)

CWL-renderable step.

Coerces the dewret structure of a step into that needed for valid CWL.

Attributes

Name	Type	Description	Default
name	None	identifier to call this step by.	None
run	None	task to execute for this step.	None
in_	None	inputs from values or other steps.	None

Static methods

from_step

def from_step(
    step: dewret.workflow.Step
) -> 'StepDefinition'

Build from a Step.

Converts a dewret.workflow.Step into a CWL-rendering object.

Parameters:

Name	Type	Description	Default
step	None	step to convert.	None

Instance variables

in_

name

out

run

Methods

render

def render(
    self
) -> dict[str, typing.Union[str, float, bool, bytes, int, NoneType, list['RawType'], dict[str, 'RawType']]]

Render to a dict-like structure.

Returns:

Type	Description
None	Reduced form as a native Python dict structure for serialization.

WorkflowDefinition

class WorkflowDefinition(
    steps: list[dewret.renderers.cwl.StepDefinition],
    inputs: dewret.renderers.cwl.InputsDefinition,
    outputs: dewret.renderers.cwl.OutputsDefinition
)

CWL-renderable workflow.

Coerces the dewret structure of a workflow into that needed for valid CWL.

Attributes

Name	Type	Description	Default
steps	None	sequence of steps in the workflow.	None

Static methods

from_workflow

def from_workflow(
    workflow: dewret.workflow.Workflow
) -> 'WorkflowDefinition'

Build from a Workflow.

Converts a dewret.workflow.Workflow into a CWL-rendering object.

Parameters:

Name	Type	Description	Default
workflow	None	workflow to convert.	None

Instance variables

inputs

outputs

steps

Methods

render

def render(
    self
) -> dict[str, typing.Union[str, float, bool, bytes, int, NoneType, list['RawType'], dict[str, 'RawType']]]

Render to a dict-like structure.

Returns:

Type	Description
None	Reduced form as a native Python dict structure for serialization.