cwltool.utils

Shared functions and other definitions.

Module Contents

Classes

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)

get_listing(fs_access, rec, recursive = True)

trim_listing(obj)

Remove 'listing' field from Directory objects that are file references.

downloadHttpFile(httpurl)

ensure_writable(path, include_root = False)

Ensure that 'path' is writable.

ensure_non_writable(path)

normalizeFilesDirs(job)

posix_path(local_path)

local_path(posix_path)

create_tmp_dir(tmpdir_prefix)

Create a temporary directory that respects the given tmpdir_prefix.

Attributes

CONTENT_LIMIT

DEFAULT_TMP_PREFIX

processes_to_kill

CWLOutputAtomType

CWLOutputType

CWLObjectType

JobsType

JobsGeneratorType

OutputCallbackType

ResolverType

DestinationsType

ScatterDestinationsType

ScatterOutputCallbackType

SinkType

DirectoryType

JSONAtomType

JSONType

WorkflowStateItem

ParametersType

StepType

fcntl

msvcrt

cwltool.utils.CONTENT_LIMIT
cwltool.utils.DEFAULT_TMP_PREFIX
cwltool.utils.processes_to_kill :Deque[subprocess.Popen[str]]
cwltool.utils.CWLOutputAtomType
cwltool.utils.CWLOutputType
cwltool.utils.CWLObjectType
cwltool.utils.JobsType
cwltool.utils.JobsGeneratorType
cwltool.utils.OutputCallbackType
cwltool.utils.ResolverType
cwltool.utils.DestinationsType
cwltool.utils.ScatterDestinationsType
cwltool.utils.ScatterOutputCallbackType
cwltool.utils.SinkType
cwltool.utils.DirectoryType
cwltool.utils.JSONAtomType
cwltool.utils.JSONType
cwltool.utils.WorkflowStateItem
cwltool.utils.ParametersType
cwltool.utils.StepType
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

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 (Union[MutableMapping[str, Any], MutableSequence[Any], Any]) –

Return type

Union[str, MutableSequence[Any], MutableMapping[str, Any]]

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

Apply a function to with “class” in cls.

Parameters
  • rec (Any) –

  • cls (Iterable[Any]) –

  • op (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 (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.fcntl :Optional[ModuleType]
cwltool.utils.msvcrt :Optional[ModuleType]
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[Callable[Ellipsis, Any], functools.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[Callable[Ellipsis, Any], functools.partial[Any]]) –

Return type

None

cwltool.utils.dedup(listing)
Parameters

listing (List[CWLObjectType]) –

Return type

List[CWLObjectType]

cwltool.utils.get_listing(fs_access, rec, recursive=True)
Parameters
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)
Parameters

httpurl (str) –

Return type

str

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)
Parameters

path (str) –

Return type

None

cwltool.utils.normalizeFilesDirs(job)
Parameters

job (Optional[Union[MutableSequence[MutableMapping[str, Any]], MutableMapping[str, Any], DirectoryType]]) –

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().

get_requirement(self, feature)
Parameters

feature (str) –

Return type

Tuple[Optional[CWLObjectType], Optional[bool]]