cwltool.utils ============= .. py:module:: cwltool.utils .. autoapi-nested-parse:: Shared functions and other definitions. Attributes ---------- .. autoapisummary:: cwltool.utils.CONTENT_LIMIT cwltool.utils.DEFAULT_TMP_PREFIX cwltool.utils.processes_to_kill cwltool.utils.JobsType cwltool.utils.JobsGeneratorType cwltool.utils.OutputCallbackType cwltool.utils.ResolverType cwltool.utils.DestinationsType cwltool.utils.ScatterDestinationsType cwltool.utils.ScatterOutputCallbackType cwltool.utils.JSONType cwltool.utils.ParametersType cwltool.utils.StepType cwltool.utils.LoadListingType Classes ------- .. autoapisummary:: cwltool.utils.DirentType cwltool.utils.WorkflowStateItem cwltool.utils.HasReqsHints Functions --------- .. autoapisummary:: cwltool.utils.versionstring cwltool.utils.aslist cwltool.utils.copytree_with_merge cwltool.utils.cmp_like_py2 cwltool.utils.bytes2str_in_dicts cwltool.utils.visit_class cwltool.utils.visit_files cwltool.utils.visit_directories cwltool.utils.visit_files_directories cwltool.utils.visit_field cwltool.utils.random_outdir cwltool.utils.shared_file_lock cwltool.utils.upgrade_lock cwltool.utils.adjustFileObjs cwltool.utils.adjustDirObjs cwltool.utils.adjustFileDirObjs cwltool.utils.dedup cwltool.utils.get_listing cwltool.utils.trim_listing cwltool.utils.downloadHttpFile cwltool.utils.ensure_writable cwltool.utils.ensure_non_writable cwltool.utils.normalizeFilesDirs cwltool.utils.posix_path cwltool.utils.local_path cwltool.utils.create_tmp_dir Module Contents --------------- .. py:data:: CONTENT_LIMIT :value: 65536 .. py:data:: DEFAULT_TMP_PREFIX .. py:data:: processes_to_kill :type: Deque[subprocess.Popen[str]] Typical raw dictionary found in lightly parsed CWL. .. py:type:: JobsType :canonical: Union['CommandLineJob', 'JobBase', 'WorkflowJob', 'ExpressionJob', 'CallbackJob'] .. py:type:: JobsGeneratorType :canonical: Generator[Optional[JobsType], None, None] .. py:type:: OutputCallbackType :canonical: Callable[[Optional[CWLObjectType], str], None] .. py:type:: ResolverType :canonical: Callable[['Loader', str], Optional[str]] .. py:type:: DestinationsType :canonical: MutableMapping[str, Optional[CWLOutputType]] .. py:type:: ScatterDestinationsType :canonical: MutableMapping[str, list[Optional[CWLOutputType]]] .. py:type:: ScatterOutputCallbackType :canonical: Callable[[Optional[ScatterDestinationsType], str], None] .. py:type:: JSONType :canonical: Union[dict[str, 'JSONType'], list['JSONType'], str, int, float, bool, None] .. py:class:: DirentType Bases: :py:obj:`TypedDict` .. autoapi-inheritance-diagram:: cwltool.utils.DirentType :parts: 1 InitialWorkDirRequirement.listing item. .. py:attribute:: entry :type: Required[str | cwl_utils.types.CWLFileType | cwl_utils.types.CWLDirectoryType] .. py:attribute:: entryname :type: str .. py:attribute:: writable :type: bool .. py:class:: WorkflowStateItem Bases: :py:obj:`NamedTuple` .. autoapi-inheritance-diagram:: cwltool.utils.WorkflowStateItem :parts: 1 Workflow state item. .. py:attribute:: parameter :type: cwl_utils.types.CWLObjectType .. py:attribute:: value :type: cwl_utils.types.CWLOutputType | None .. py:attribute:: success :type: str .. py:type:: ParametersType :canonical: list[CWLObjectType] .. py:type:: StepType :canonical: CWLObjectType .. py:data:: LoadListingType .. py:function:: versionstring() Version of CWLtool used to execute the workflow. .. py:function:: aslist(thing) Wrap any non-MutableSequence/list in a list. .. py:function:: copytree_with_merge(src, dst) .. py:function:: 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` .. py:function:: bytes2str_in_dicts(inp) Convert any present byte string to unicode string, inplace. input is a dict of nested dicts and lists .. py:function:: visit_class(rec, cls, op) Apply a function to with "class" in cls. .. py:function:: visit_files(rec, op) Apply a function for any File objects. .. py:function:: visit_directories(rec, op) Apply a function for any Directory objects. .. py:function:: visit_files_directories(rec, op) Apply a function for any File or Directory objects. .. py:function:: visit_field(rec, field, op) Apply a function to mapping with 'field'. .. py:function:: random_outdir() Return the random directory name chosen to use for tool / workflow output. .. py:function:: shared_file_lock(fd) .. py:function:: upgrade_lock(fd) .. py:function:: adjustFileObjs(rec, op) Apply an update function to each File object in the object `rec`. .. py:function:: adjustDirObjs(rec, op) Apply an update function to each Directory object in the object `rec`. .. py:function:: adjustFileDirObjs(rec, op) Apply an update function to each File/Directory object in the object `rec`. .. py:function:: dedup(listing) Remove duplicate entries from a CWL Directory 'listing'. .. py:function:: get_listing(fs_access, rec, recursive = True) Expand, recursively, any 'listing' fields in a Directory. .. py:function:: 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. .. py:function:: 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. .. py:function:: 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. .. py:function:: ensure_non_writable(path) Attempt to change permissions to ensure that a path is not writable. .. py:function:: 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 .. py:function:: posix_path(local_path) .. py:function:: local_path(posix_path) .. py:function:: create_tmp_dir(tmpdir_prefix) Create a temporary directory that respects the given tmpdir_prefix. .. py:class:: HasReqsHints Base class for get_requirement(). .. py:attribute:: requirements :type: list[cwl_utils.types.CWLObjectType] :value: [] .. py:attribute:: hints :type: list[cwl_utils.types.CWLObjectType] :value: [] .. py:method:: get_requirement(feature) Retrieve the named feature from the requirements field, or the hints field.