cwl_utils.utils

Miscellaneous utility functions.

Attributes

fast_yaml

Functions

bytes2str_in_dicts(inp)

Convert any present byte string to unicode string, inplace.

load_linked_file(base_url, link[, is_import])

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

normalize_to_map(obj, key_field)

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

normalize_to_list(obj, key_field, value_field)

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

resolved_path(base_url, link)

Derive a resolved path.

singularity_supports_userns()

Confirm if the version of Singularity install supports the --userns flag.

yaml_dumps(obj)

Shortcut.

to_pascal_case(name)

Convert a string to PascalCase.

sanitise_schema_field(schema_field_item)

Schemas need to be resolved before converted to JSON properties.

is_uri(uri)

Given a URI return True if it is a URI.

is_local_uri(uri)

Given a uri, first check if it is a uri, then check if it is a local uri.

get_value_from_uri(uri)

Given a URI, return the value after #.

Module Contents

cwl_utils.utils.fast_yaml
cwl_utils.utils.bytes2str_in_dicts(inp)

Convert any present byte string to unicode string, inplace.

input is a dict of nested dicts and lists

Parameters:

inp (Union[collections.abc.MutableMapping[str, Any], collections.abc.MutableSequence[Any], Any])

Return type:

Union[str, collections.abc.MutableSequence[Any], collections.abc.MutableMapping[str, Any]]

cwl_utils.utils.load_linked_file(base_url, link, is_import=False)

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

Parameters:
Return type:

tuple[Any, urllib.parse.ParseResult]

cwl_utils.utils.normalize_to_map(obj, key_field)

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

Parameters:
Return type:

dict[str, Any]

cwl_utils.utils.normalize_to_list(obj, key_field, value_field)

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

Parameters:
Return type:

list[Any]

cwl_utils.utils.resolved_path(base_url, link)

Derive a resolved path.

This function will 1. Resolve the path, which means dot and double dot components are resolved 2. Use the OS appropriate path resolution for local paths, and network appropriate resolution for network paths

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

Parameters:
Returns:

new URL that allows us to retrieve the linked document

Return type:

urllib.parse.ParseResult

cwl_utils.utils.singularity_supports_userns()

Confirm if the version of Singularity install supports the –userns flag.

Return type:

bool

cwl_utils.utils.yaml_dumps(obj)

Shortcut.

Don’t use if you have a file descriptor (like sys.stdout) available.

Parameters:

obj (Any)

Return type:

str

cwl_utils.utils.to_pascal_case(name)

Convert a string to PascalCase.

fastq-list-row to FastqListRow fastq_list_row to FastqListRow :param name: :return:

Parameters:

name (str)

Return type:

str

cwl_utils.utils.sanitise_schema_field(schema_field_item)

Schemas need to be resolved before converted to JSON properties.

Convert
{

‘type’: ‘Directory?’

}

To
{

‘type’: [‘null’, ‘Directory’]

}

Convert
{

‘type’: ‘string[]’

}

To
InputArraySchema(

type_=array, items=string

)

Convert
{

‘type’: ‘File[]?’

}

To
{
‘type’: [
‘null’, InputArraySchema(

type_=array, items=File

)

]

}

Convert
{

‘type’: ‘Enum’, ‘symbols’: [‘A’, ‘B’, ‘C’]

}

To
{
‘type’: InputEnumSchema(

type_=enum, symbols=[‘A’, ‘B’, ‘C’]

)

}

Convert
{

‘type’: ‘array’, ‘items’: {

‘$import’: ‘../../../schemas/fastq-list-row/1.0.0/fastq-list-row__1.0.0.yaml#fastq-list-row’

}

}

To
{
‘type’: InputArraySchema(

type_=array, items={

‘$import’: ‘../../../schemas/fastq-list-row/1.0.0/fastq-list-row__1.0.0.yaml#fastq-list-row’

}

)

}

Parameters:

schema_field_item (Union[Dict[str, Any], str])

Returns:

Return type:

Union[Dict[str, Any], str]

cwl_utils.utils.is_uri(uri)

Given a URI return True if it is a URI.

Parameters:

uri (str)

Returns:

Return type:

bool

cwl_utils.utils.is_local_uri(uri)

Given a uri, first check if it is a uri, then check if it is a local uri.

Parameters:

uri (str)

Return type:

bool

cwl_utils.utils.get_value_from_uri(uri)

Given a URI, return the value after #.

file://path/to/imported/record#my_workflow_name/record_name Returns record_name :param uri: :return:

Parameters:

uri (str)

Return type:

str