cwl_utils.pack

CWL document packing functions.

The link resolution is as follows:

We always have two components: the base and the link If the link is a url or absolute path it is what is used to fetch the data. If the link is a relative path it is combined with the base and that is what is used to fetch data

From https://github.com/rabix/sbpack/blob/b8404a0859ffcbe1edae6d8f934e51847b003320/sbpack/lib.py

Attributes

logger

Functions

`get_inner_dict`(cwl, path)
`pack_process`(cwl, base_url, cwl_version[, ...])
`listify_everything`(cwl)	Convert many CWL construct from their map to the list version.
`normalize_sources`(cwl)	Normalize the steps and output of a CWL Workflow.
`load_schemadefs`(cwl, base_url[, parent_user_defined_types])	Internalize any SchemaDefRequirement, and remove it.
`resolve_schemadefs`(cwl, base_url, user_defined_types)
`resolve_imports`(cwl, base_url)
`resolve_steps`(cwl, base_url, cwl_version[, ...])	Load and pack all "run" sections of the workflow steps.
`add_missing_requirements`(cwl)	Due to packing, we may need to add a "SubworkflowFeatureRequirement".
`pack`(cwl_path)	Pack a CWL document at the given path.

Module Contents

cwl_utils.pack.logger

cwl_utils.pack.get_inner_dict(cwl, path)

Parameters:

cwl (dict[str, Any])
path (list[dict[str, Any]])

Return type:

Optional[dict[str, Any]]

cwl_utils.pack.pack_process(cwl, base_url, cwl_version, parent_user_defined_types=None)

Parameters:

cwl (dict[str, Any])
base_url (urllib.parse.ParseResult)
cwl_version (str)
parent_user_defined_types (Optional[dict[str, Any]])

Return type:

dict[str, Any]

cwl_utils.pack.listify_everything(cwl)

Convert many CWL construct from their map to the list version.

See https://www.commonwl.org/v1.1/Workflow.html#map

Parameters:: cwl (dict[str, Any])
Return type:: dict[str, Any]

cwl_utils.pack.normalize_sources(cwl)

Normalize the steps and output of a CWL Workflow.

Parameters:: cwl (dict[str, Any])
Return type:: dict[str, Any]

cwl_utils.pack.load_schemadefs(cwl, base_url, parent_user_defined_types=None)

Internalize any SchemaDefRequirement, and remove it.

Parameters:

cwl (dict[str, Any])
base_url (urllib.parse.ParseResult)
parent_user_defined_types (Optional[dict[str, Any]])

Return type:

tuple[dict[str, Any], dict[str, Any]]

cwl_utils.pack.resolve_schemadefs(cwl, base_url, user_defined_types)

Parameters:

cwl (dict[str, Any])
base_url (urllib.parse.ParseResult)
user_defined_types (dict[str, Any])

Return type:

dict[str, Any]

cwl_utils.pack.resolve_imports(cwl, base_url)

Parameters:

cwl (Any)
base_url (urllib.parse.ParseResult)

Return type:

Any

cwl_utils.pack.resolve_steps(cwl, base_url, cwl_version, parent_user_defined_types=None)

Load and pack all “run” sections of the workflow steps.

Parameters:

cwl (dict[str, Any])
base_url (urllib.parse.ParseResult)
cwl_version (str)
parent_user_defined_types (Optional[dict[str, Any]])

Return type:

dict[str, Any]

cwl_utils.pack.add_missing_requirements(cwl)

Due to packing, we may need to add a “SubworkflowFeatureRequirement”.

Parameters:: cwl (dict[str, Any])
Return type:: dict[str, Any]

cwl_utils.pack.pack(cwl_path)

Pack a CWL document at the given path.

Parameters:: cwl_path (str)
Return type:: dict[str, Any]