Module dewret.renderers.cwl

CWL Renderer.

Outputs a Common Workflow Language representation of the current workflow.

Variables

InputSchemaType

Functions

cwl_type_from_value

def cwl_type_from_value(
    label: str,
    val: str | float | bool | bytes | int | None | list['RawType'] | dict[str, 'RawType'] | dewret.utils.Unset
) -> dewret.renderers.cwl.CommandInputSchema

Find a CWL type for a given (possibly Unset) value.

Parameters:

Name	Type	Description	Default
label	None	the label for the variable being checked to prefill the input def and improve debugging info.	None
val	None	a raw Python variable or an unset variable.	None

Returns:

Type	Description
None	Input schema type.

default_config

def default_config(
    
) -> dewret.renderers.cwl.CWLRendererConfiguration

Default configuration for this renderer.

This is a hook-like call to give a configuration dict that this renderer will respect, and sets any necessary default values.

Returns: a dict with (preferably) raw type structures to enable easy setting from YAML/JSON.

raw_to_command_input_schema

def raw_to_command_input_schema(
    label: str,
    value: str | float | bool | bytes | int | None | list['RawType'] | dict[str, 'RawType'] | dewret.utils.Unset
) -> str | dewret.renderers.cwl.CommandInputSchema | list[str] | list['InputSchemaType'] | dict[str, 'str | 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,
    **kwargs: *<class 'dewret.renderers.cwl.CWLRendererConfiguration'>
) -> dict[str, dict[str, str | float | bool | bytes | int | None | list['RawType'] | dict[str, 'RawType']]]

Render to a dict-like structure.

Parameters:

Name	Type	Description	Default
workflow	None	workflow to evaluate result.	None
**kwargs	None	additional configuration arguments - these should match CWLRendererConfiguration.	None

Returns:

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

render_expression

def render_expression(
    ref: Any
) -> 'ReferenceDefinition'

Turn a rich (sympy) expression into a CWL JS expression.

Parameters:

Name	Type	Description	Default
ref	None	a structure whose elements are all string-renderable or sympy Basic.	None

to_cwl_type

def to_cwl_type(
    label: str,
    typ: Type[~T]
) -> dewret.renderers.cwl.CommandInputSchema

Map Python types to CWL types.

Parameters:

Name	Type	Description	Default
label	None	the label for the variable being checked to prefill the input def and improve debugging info.	None
typ	None	a Python basic type.	None

Returns:

Type	Description
None	CWL specification type dict.

to_name

def to_name(
    result: dewret.core.Reference[typing.Any]
) -> str

Take a reference and get a name representing it.

The primary purpose of this method is to deal with the case where a reference is to the whole result, as we always put this into an imagined out field for CWL consistency.

Returns: the name of the reference, including any field portion, appending an "out" fieldname if none.

to_output_schema

def to_output_schema(
    label: str,
    typ: type[str | float | bool | bytes | int | None | 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

with_field

def with_field(
    result: Any
) -> str

Get a string representing any 'field' suffix of a value.

This only makes sense in the context of a Reference, which can represent a deep reference with a known variable (parameter or step result, say) using its __field__ attribute. Defaults to "out" as this produces compliant CWL where every output has a "fieldname".

Returns: a string representation of the field portion of the passed value or "out".

with_type

def with_type(
    result: Any
) -> type | typing.Any

Get a Python type from a value.

Does so either by using its __type__ field (for example, for References) or if unavailable, using type().

Returns: a Python type.

Classes

CWLRendererConfiguration

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

Configuration for the renderer.

Attributes

Name	Type	Description	Default
allow_complex_types	None	can input/output types be other than raw?	None
factories_as_params	None	should factories be treated as input or steps?	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

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[typing.Union[dewret.workflow.ParameterReference[typing.Any], dewret.workflow.FactoryCall]]
) -> '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, str | float | bool | bytes | int | None | 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'] | list['CommandOutputSchema'] | dewret.renderers.cwl.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]] | list[dewret.workflow.StepReference[typing.Any]] | tuple[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, str | float | bool | bytes | int | None | list['RawType'] | dict[str, 'RawType']] | list[str | float | bool | bytes | int | None | 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 | None,
    value_from: str | None
)

CWL-renderable internal reference.

Normally points to a value or a step.

Static methods

from_reference

def from_reference(
    ref: dewret.core.Reference[typing.Any]
) -> '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

value_from

Methods

render

def render(
    self
) -> dict[str, str | float | bool | bytes | int | None | 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.core.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.BaseStep
) -> 'StepDefinition'

Build from a BaseStep.

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, str | float | bool | bytes | int | None | 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,
    name: None | str
)

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,
    name: None | str = None
) -> '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
name	None	name of this workflow, if it should have one.	None

Instance variables

inputs

name

outputs

steps

Methods

render

def render(
    self
) -> dict[str, str | float | bool | bytes | int | None | 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.