# cwltool.utils

Shared functions and other definitions.

## Attributes

`CONTENT_LIMIT` |
`DEFAULT_TMP_PREFIX` |
`processes_to_kill` |
`CWLOutputType` |
`CWLObjectType` | Typical raw dictionary found in lightly parsed CWL.
`JobsType` |
`JobsGeneratorType` |
`OutputCallbackType` |
`ResolverType` |
`DestinationsType` |
`ScatterDestinationsType` |
`ScatterOutputCallbackType` |
`SinkType` |
`DirectoryType` |
`JSONType` |
`ParametersType` |
`StepType` |
`LoadListingType` |

## Classes

`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) |
`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) |
`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]]

type cwltool.utils.CWLOutputType = None | bool | str | int | float | MutableSequence['CWLOutputType'] | MutableMapping[str, 'CWLOutputType']

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

  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.SinkType = CWLOutputType | CWLObjectType

cwltool.utils.DirectoryType

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

class cwltool.utils.WorkflowStateItem

  Bases: `NamedTuple`

  Workflow state item.

  parameter: CWLObjectType
  value: 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)
  Parameters:

    listing (list[CWLObjectType])

  Return type:

    list[CWLObjectType]

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

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

  Parameters:

    - fs_access (cwltool.stdfsaccess.StdFsAccess)

    - rec (CWLObjectType)

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

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

  requirements: list[CWLObjectType] = []
  hints: list[CWLObjectType] = []
  get_requirement(feature)

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

    Parameters:

      feature (str)

    Return type:

      tuple[CWLObjectType | None, bool | None]