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:

dict[str, Any] | None

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 (dict[str, Any] | None)

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 (dict[str, Any] | None)

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 (dict[str, Any] | None)

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]