cwltool.utils

Shared functions and other definitions.

Attributes

CONTENT_LIMIT
DEFAULT_TMP_PREFIX
processes_to_kill	Typical raw dictionary found in lightly parsed CWL.
JobsType
JobsGeneratorType
OutputCallbackType
ResolverType
DestinationsType
ScatterDestinationsType
ScatterOutputCallbackType
JSONType
ParametersType
StepType
LoadListingType

Classes

DirentType	InitialWorkDirRequirement.listing item.
WorkflowStateItem	Workflow state item.
HasReqsHints	Base class for get_requirement().

Functions

versionstring()	Version of CWLtool used to execute the workflow.
aslist(thing)	Wrap any non-MutableSequence/list in a list.
copytree_with_merge(src, dst)
cmp_like_py2(dict1, dict2)	Compare in the same manner as Python2.
bytes2str_in_dicts(inp)	Convert any present byte string to unicode string, inplace.
visit_class(rec, cls, op)	Apply a function to with "class" in cls.
visit_field(rec, field, op)	Apply a function to mapping with 'field'.
random_outdir()	Return the random directory name chosen to use for tool / workflow output.
shared_file_lock(fd)
upgrade_lock(fd)
adjustFileObjs(rec, op)	Apply an update function to each File object in the object rec.
adjustDirObjs(rec, op)	Apply an update function to each Directory object in the object rec.
dedup(listing)	Remove duplicate entries from a CWL Directory 'listing'.
get_listing(fs_access, rec[, recursive])	Expand, recursively, any 'listing' fields in a Directory.
trim_listing(obj)	Remove 'listing' field from Directory objects that are file references.
downloadHttpFile(httpurl)	Download a remote file, possibly using a locally cached copy.
ensure_writable(path[, include_root])	Ensure that 'path' is writable.
ensure_non_writable(path)	Attempt to change permissions to ensure that a path is not writable.
normalizeFilesDirs(job)	Add missing `location`s and `basename`s to CWL File and Directory objects.
posix_path(local_path)
local_path(posix_path)
create_tmp_dir(tmpdir_prefix)	Create a temporary directory that respects the given tmpdir_prefix.

Module Contents

cwltool.utils.CONTENT_LIMIT = 65536

cwltool.utils.DEFAULT_TMP_PREFIX

cwltool.utils.processes_to_kill: Deque[subprocess.Popen[str]]

Typical raw dictionary found in lightly parsed CWL.

type cwltool.utils.JobsType = 'CommandLineJob' | 'JobBase' | 'WorkflowJob' | 'ExpressionJob' | 'CallbackJob'

type cwltool.utils.JobsGeneratorType = Generator[JobsType | None, None, None]

type cwltool.utils.OutputCallbackType = Callable[[CWLObjectType | None, str], None]

type cwltool.utils.ResolverType = Callable[['Loader', str], str | None]

type cwltool.utils.DestinationsType = MutableMapping[str, CWLOutputType | None]

type cwltool.utils.ScatterDestinationsType = MutableMapping[str, list[CWLOutputType | None]]

type cwltool.utils.ScatterOutputCallbackType = Callable[[ScatterDestinationsType | None, str], None]

type cwltool.utils.JSONType = dict[str, 'JSONType'] | list['JSONType'] | str | int | float | bool | None

class cwltool.utils.DirentType

Bases: TypedDict

InitialWorkDirRequirement.listing item.

entry: Required[str | cwl_utils.types.CWLFileType | cwl_utils.types.CWLDirectoryType]

entryname: str

writable: bool

class cwltool.utils.WorkflowStateItem

Bases: NamedTuple

Workflow state item.

parameter: cwl_utils.types.CWLObjectType

value: cwl_utils.types.CWLOutputType | None

success: str

type cwltool.utils.ParametersType = list[CWLObjectType]

type cwltool.utils.StepType = CWLObjectType

cwltool.utils.LoadListingType

cwltool.utils.versionstring()

Version of CWLtool used to execute the workflow.

Return type:

str

cwltool.utils.aslist(thing)

Wrap any non-MutableSequence/list in a list.

Parameters:

thing (Any)

Return type:

collections.abc.MutableSequence[Any]

cwltool.utils.copytree_with_merge(src, dst)

Parameters:

• src (str)

• dst (str)

Return type:

None

cwltool.utils.cmp_like_py2(dict1, dict2)

Compare in the same manner as Python2.

Comparison function to be used in sorting as python3 doesn’t allow sorting of different types like str() and int(). This function re-creates sorting nature in py2 of heterogeneous list of int and str

Parameters:

• dict1 (dict[str, Any])

• dict2 (dict[str, Any])

Return type:

int

cwltool.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 (collections.abc.MutableMapping[str, Any] | collections.abc.MutableSequence[Any] | Any)

Return type:

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

cwltool.utils.visit_class(rec, cls, op)

Apply a function to with “class” in cls.

Parameters:

• rec (Any)

• cls (collections.abc.Iterable[Any])

• op (collections.abc.Callable[Ellipsis, Any])

Return type:

None

cwltool.utils.visit_field(rec, field, op)

Apply a function to mapping with ‘field’.

Parameters:

• rec (Any)

• field (str)

• op (collections.abc.Callable[Ellipsis, Any])

Return type:

None

cwltool.utils.random_outdir()

Return the random directory name chosen to use for tool / workflow output.

Return type:

str

cwltool.utils.shared_file_lock(fd)

Parameters:

fd (IO[Any])

Return type:

None

cwltool.utils.upgrade_lock(fd)

Parameters:

fd (IO[Any])

Return type:

None

cwltool.utils.adjustFileObjs(rec, op)

Apply an update function to each File object in the object rec.

Parameters:

• rec (Any)

• op (Union[collections.abc.Callable[Ellipsis, Any], partial[Any]])

Return type:

None

cwltool.utils.adjustDirObjs(rec, op)

Apply an update function to each Directory object in the object rec.

Parameters:

• rec (Any)

• op (Union[collections.abc.Callable[Ellipsis, Any], partial[Any]])

Return type:

None

cwltool.utils.dedup(listing)

Remove duplicate entries from a CWL Directory ‘listing’.

Parameters:

listing (collections.abc.MutableSequence[cwl_utils.types.CWLFileType | cwl_utils.types.CWLDirectoryType])

Return type:

collections.abc.MutableSequence[cwl_utils.types.CWLFileType | cwl_utils.types.CWLDirectoryType]

cwltool.utils.get_listing(fs_access, rec, recursive=True)

Expand, recursively, any ‘listing’ fields in a Directory.

Parameters:

• fs_access (cwltool.stdfsaccess.StdFsAccess)

• rec (cwl_utils.types.CWLObjectType | cwl_utils.types.CWLDirectoryType)

• recursive (bool)

Return type:

None

cwltool.utils.trim_listing(obj)

Remove ‘listing’ field from Directory objects that are file references.

It redundant and potentially expensive to pass fully enumerated Directory objects around if not explicitly needed, so delete the ‘listing’ field when it is safe to do so.

Parameters:

obj (dict[str, Any])

Return type:

None

cwltool.utils.downloadHttpFile(httpurl)

Download a remote file, possibly using a locally cached copy.

Returns a tuple: - the local path for the downloaded file - the Last-Modified timestamp if received from the remote server.

Parameters:

httpurl (str)

Return type:

tuple[str, datetime.datetime | None]

cwltool.utils.ensure_writable(path, include_root=False)

Ensure that ‘path’ is writable.

If ‘path’ is a directory, then all files and directories under ‘path’ are made writable, recursively. If ‘path’ is a file or if ‘include_root’ is True, then ‘path’ itself is made writable.

Parameters:

• path (str)

• include_root (bool)

Return type:

None

cwltool.utils.ensure_non_writable(path)

Attempt to change permissions to ensure that a path is not writable.

Parameters:

path (str)

Return type:

None

cwltool.utils.normalizeFilesDirs(job)

Add missing `location`s and `basename`s to CWL File and Directory objects.

Raises:

ValidationException – if anonymous objects are missing required fields, or if the location ends in ‘/’ but the object isn’t a directory

Parameters:

job (collections.abc.Sequence[cwl_utils.types.CWLObjectType | cwl_utils.types.CWLOutputType | None] | cwl_utils.types.CWLObjectType | cwl_utils.types.CWLOutputType | None)

Return type:

None

cwltool.utils.posix_path(local_path)

Parameters:

local_path (str)

Return type:

str

cwltool.utils.local_path(posix_path)

Parameters:

posix_path (str)

Return type:

str

cwltool.utils.create_tmp_dir(tmpdir_prefix)

Create a temporary directory that respects the given tmpdir_prefix.

Parameters:

tmpdir_prefix (str)

Return type:

str

class cwltool.utils.HasReqsHints

Base class for get_requirement().

requirements: list[cwl_utils.types.CWLObjectType] = []

hints: list[cwl_utils.types.CWLObjectType] = []

get_requirement(feature)

Retrieve the named feature from the requirements field, or the hints field.

Parameters:

feature (str)

Return type:

tuple[cwl_utils.types.CWLObjectType | None, bool | None]