Source code for toil.cwl.utils

# Copyright (C) 2015-2021 Regents of the University of California
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.

"""Utility functions used for Toil's CWL interpreter."""

import logging
import os
from pathlib import PurePosixPath
import posixpath
import stat
from typing import (

from toil.fileStores import FileID
from toil.fileStores.abstractFileStore import AbstractFileStore
from toil.jobStores.abstractJobStore import AbstractJobStore

logger = logging.getLogger(__name__)

# Customized CWL utilities

# What exit code do we need to bail with if we or any of the local jobs that
# parse workflow files see an unsupported feature?

# And what error will make the worker exit with that code

[docs] class CWLUnsupportedException(Exception): """Fallback exception."""
try: import cwltool.errors CWL_UNSUPPORTED_REQUIREMENT_EXCEPTION: Union[ Type[cwltool.errors.UnsupportedRequirement], Type[CWLUnsupportedException] ] = cwltool.errors.UnsupportedRequirement except ImportError: CWL_UNSUPPORTED_REQUIREMENT_EXCEPTION = CWLUnsupportedException
[docs] def visit_top_cwl_class( rec: Any, classes: Iterable[str], op: Callable[[Any], Any] ) -> None: """ Apply the given operation to all top-level CWL objects with the given named CWL class. Like cwltool's visit_class but doesn't look inside any object visited. """ if isinstance(rec, MutableMapping): if rec.get("class", None) in classes: # This is one of the classes requested # So process it op(rec) else: # Look inside it instead for key in rec: visit_top_cwl_class(rec[key], classes, op) elif isinstance(rec, MutableSequence): # This item is actually a list of things, so look at all of them. for key in rec: visit_top_cwl_class(key, classes, op)
DownReturnType = TypeVar("DownReturnType") UpReturnType = TypeVar("UpReturnType")
[docs] def visit_cwl_class_and_reduce( rec: Any, classes: Iterable[str], op_down: Callable[[Any], DownReturnType], op_up: Callable[[Any, DownReturnType, List[UpReturnType]], UpReturnType], ) -> List[UpReturnType]: """ Apply the given operations to all CWL objects with the given named CWL class. Applies the down operation top-down, and the up operation bottom-up, and passes the down operation's result and a list of the up operation results for all child keys (flattening across lists and collapsing nodes of non-matching classes) to the up operation. :returns: The flattened list of up operation results from all calls. """ results = [] if isinstance(rec, MutableMapping): child_results = [] if rec.get("class", None) in classes: # Apply the down operation down_result = op_down(rec) for key in rec: # Look inside and collect child results for result in visit_cwl_class_and_reduce(rec[key], classes, op_down, op_up): child_results.append(result) if rec.get("class", None) in classes: # Apply the up operation results.append(op_up(rec, down_result, child_results)) else: # We aren't processing here so pass up all the child results results += child_results elif isinstance(rec, MutableSequence): # This item is actually a list of things, so look at all of them. for key in rec: for result in visit_cwl_class_and_reduce(key, classes, op_down, op_up): # And flatten together all their results. results.append(result) return results
DirectoryStructure = Dict[str, Union[str, "DirectoryStructure"]]
[docs] def get_from_structure(dir_dict: DirectoryStructure, path: str) -> Union[str, DirectoryStructure, None]: """ Given a relative path, follow it in the given directory structure. Return the string URI for files, the directory dict for subdirectories, or None for nonexistent things. """ # Resolve .. and split into path components parts = PurePosixPath(posixpath.normpath(path)).parts if len(parts) == 0: return dir_dict if parts[0] in ('..', '/'): raise RuntimeError(f"Path {path} not resolvable in virtual directory") found: Union[str, DirectoryStructure] = dir_dict for part in parts: # Go down by each path component in turn if isinstance(found, str): # Looking for a subdirectory of a file, which doesn't exist return None if part not in found: return None found = found[part] # Now we're at the place we want to be. return found
[docs] def download_structure( file_store: AbstractFileStore, index: Dict[str, str], existing: Dict[str, str], dir_dict: DirectoryStructure, into_dir: str, ) -> None: """ Download nested dictionary from the Toil file store to a local path. Guaranteed to fill the structure with real files, and not symlinks out of it to elsewhere. File URIs may be toilfile: URIs or any other URI that Toil's job store system can read. :param file_store: The Toil file store to download from. :param index: Maps from downloaded file path back to input URI. :param existing: Maps from file_store_id URI to downloaded file path. :param dir_dict: a dict from string to string (for files) or dict (for subdirectories) describing a directory structure. :param into_dir: The directory to download the top-level dict's files into. """ logger.debug("Downloading directory with %s items", len(dir_dict)) for name, value in dir_dict.items(): if name == ".": # Skip this key that isn't a real child file. continue if isinstance(value, dict): # This is a subdirectory, so make it and download # its contents logger.debug("Downloading subdirectory '%s'", name) subdir = os.path.join(into_dir, name) os.mkdir(subdir) download_structure(file_store, index, existing, value, subdir) else: # This must be a file path uploaded to Toil. if not isinstance(value, str): raise RuntimeError(f"Did not find a file at {value}.") logger.debug("Downloading contained file '%s'", name) dest_path = os.path.join(into_dir, name) if value.startswith("toilfile:"): # So download the file into place. # Make sure to get a real copy of the file because we may need to # mount the directory into a container as a whole. file_store.readGlobalFile( FileID.unpack(value[len("toilfile:") :]), dest_path, symlink=False ) else: # We need to download from some other kind of URL. size, executable = AbstractJobStore.read_from_url(value, open(dest_path, 'wb')) if executable: # Make the written file executable os.chmod(dest_path, os.stat(dest_path).st_mode | stat.S_IXUSR) # Update the index dicts # TODO: why? index[dest_path] = value existing[value] = dest_path