cwltool.process

Classes and methods relevant for all CWL Process types.

Attributes

supportedProcessRequirements
cwl_files
salad_files
SCHEMA_CACHE
SCHEMA_FILE
SCHEMA_DIR
SCHEMA_ANY
custom_schemas
FILE_COUNT_WARNING
CWL_IANA

Classes

LogAsDebugFilter	Filter instances are used to perform arbitrary filtering of LogRecords.
Process	Abstract CWL Process.

Functions

use_standard_schema(version)
use_custom_schema(version, name, text)
get_schema(version)
shortname(inputid)
stage_files(pathmapper[, stage_func, ignore_writable, ...])	Link or copy files to their targets. Create them as needed.
relocateOutputs(outputObj, destination_path, ...[, ...])
cleanIntermediate(output_dirs)
add_sizes(fsaccess, obj)
fill_in_defaults(inputs, job, fsaccess)	For each missing input in the input object, copy over the default.
avroize_type(field_type[, name_prefix])	Add missing information to a type so that CWL types are valid.
get_overrides(overrides, toolid)	Combine overrides for the target tool ID.
var_spool_cwl_detector(obj[, item, obj_key])	Detect any textual reference to /var/spool/cwl.
eval_resource(builder, resource_req)
uniquename(stem[, names])	Construct a thread-unique name using the given stem as a prefix.
nestdir(base, deps)
mergedirs(listing)
scandeps(base, doc, reffields, urlfields, loadref[, ...])	Search for external files references in a CWL document or input object.
compute_checksums(fs_access, fileobj)

Module Contents

class cwltool.process.LogAsDebugFilter(name, parent)

Bases: logging.Filter

Filter instances are used to perform arbitrary filtering of LogRecords.

Loggers and Handlers can optionally use Filter instances to filter records as desired. The base filter class only allows events which are below a certain point in the logger hierarchy. For example, a filter initialized with “A.B” will allow events logged by loggers “A.B”, “A.B.C”, “A.B.C.D”, “A.B.D” etc. but not “A.BB”, “B.A.B” etc. If initialized with the empty string, all events are passed.

Parameters:

• name (str)

• parent (logging.Logger)

parent

filter(record)

Determine if the specified record is to be logged.

Returns True if the record should be logged, or False otherwise. If deemed appropriate, the record may be modified in-place.

Parameters:

record (logging.LogRecord)

Return type:

bool

cwltool.process.supportedProcessRequirements

cwltool.process.cwl_files = ('Base.yml', 'Workflow.yml', 'CommandLineTool.yml', 'CommonWorkflowLanguage.yml', 'Process.yml',...

cwltool.process.salad_files = ('metaschema.yml', 'metaschema_base.yml', 'salad.md', 'field_name.yml', 'import_include.md',...

cwltool.process.SCHEMA_CACHE: dict[str, tuple[schema_salad.ref_resolver.Loader, schema_salad.avro.schema.Names | schema_salad.avro.schema.SchemaParseException, cwltool.utils.CWLObjectType, schema_salad.ref_resolver.Loader]]

cwltool.process.SCHEMA_FILE: cwltool.utils.CWLObjectType | None = None

cwltool.process.SCHEMA_DIR: cwltool.utils.CWLObjectType | None = None

cwltool.process.SCHEMA_ANY: cwltool.utils.CWLObjectType | None = None

cwltool.process.custom_schemas: dict[str, tuple[str, str]]

cwltool.process.use_standard_schema(version)

Parameters:

version (str)

Return type:

None

cwltool.process.use_custom_schema(version, name, text)

Parameters:

• version (str)

• name (str)

• text (str)

Return type:

None

cwltool.process.get_schema(version)

Parameters:

version (str)

Return type:

tuple[schema_salad.ref_resolver.Loader, Union[schema_salad.avro.schema.Names, schema_salad.avro.schema.SchemaParseException], cwltool.utils.CWLObjectType, schema_salad.ref_resolver.Loader]

cwltool.process.shortname(inputid)

Parameters:

inputid (str)

Return type:

str

cwltool.process.stage_files(pathmapper, stage_func=None, ignore_writable=False, symlink=True, secret_store=None, fix_conflicts=False)

Link or copy files to their targets. Create them as needed.

Raises:

WorkflowException – if there is a file staging conflict

Parameters:

• pathmapper (cwltool.pathmapper.PathMapper)

• stage_func (Optional[Callable[[str, str], None]])

• ignore_writable (bool)

• symlink (bool)

• secret_store (Optional[cwltool.secrets.SecretStore])

• fix_conflicts (bool)

Return type:

None

cwltool.process.relocateOutputs(outputObj, destination_path, source_directories, action, fs_access, compute_checksum=True, path_mapper=PathMapper)

Parameters:

• outputObj (cwltool.utils.CWLObjectType)

• destination_path (str)

• source_directories (set[str])

• action (str)

• fs_access (cwltool.stdfsaccess.StdFsAccess)

• compute_checksum (bool)

• path_mapper (type[cwltool.pathmapper.PathMapper])

Return type:

cwltool.utils.CWLObjectType

cwltool.process.cleanIntermediate(output_dirs)

Parameters:

output_dirs (collections.abc.Iterable[str])

Return type:

None

cwltool.process.add_sizes(fsaccess, obj)

Parameters:

• fsaccess (cwltool.stdfsaccess.StdFsAccess)

• obj (cwltool.utils.CWLObjectType)

Return type:

None

cwltool.process.fill_in_defaults(inputs, job, fsaccess)

For each missing input in the input object, copy over the default.

Raises:

WorkflowException – if a required input parameter is missing

Parameters:

• inputs (list[cwltool.utils.CWLObjectType])

• job (cwltool.utils.CWLObjectType)

• fsaccess (cwltool.stdfsaccess.StdFsAccess)

Return type:

None

cwltool.process.avroize_type(field_type, name_prefix='')

Add missing information to a type so that CWL types are valid.

Parameters:

• field_type (Union[cwltool.utils.CWLObjectType, collections.abc.MutableSequence[Any], cwltool.utils.CWLOutputType, None])

• name_prefix (str)

Return type:

Union[cwltool.utils.CWLObjectType, collections.abc.MutableSequence[Any], cwltool.utils.CWLOutputType, None]

cwltool.process.get_overrides(overrides, toolid)

Combine overrides for the target tool ID.

Parameters:

• overrides (collections.abc.MutableSequence[cwltool.utils.CWLObjectType])

• toolid (str)

Return type:

cwltool.utils.CWLObjectType

cwltool.process.var_spool_cwl_detector(obj, item=None, obj_key=None)

Detect any textual reference to /var/spool/cwl.

Parameters:

• obj (cwltool.utils.CWLOutputType)

• item (Optional[Any])

• obj_key (Optional[Any])

Return type:

bool

cwltool.process.eval_resource(builder, resource_req)

Parameters:

• builder (cwltool.builder.Builder)

• resource_req (Union[str, int, float])

Return type:

Optional[Union[str, int, float]]

cwltool.process.FILE_COUNT_WARNING = 5000

class cwltool.process.Process(toolpath_object, loadingContext)

Bases: cwltool.utils.HasReqsHints

Abstract CWL Process.

Parameters:

• toolpath_object (ruamel.yaml.comments.CommentedMap)

• loadingContext (cwltool.context.LoadingContext)

metadata: cwltool.utils.CWLObjectType

provenance_object: cwltool.cwlprov.provenance_profile.ProvenanceProfile | None = None

parent_wf: cwltool.cwlprov.provenance_profile.ProvenanceProfile | None = None

names

tool

requirements

hints

original_requirements

original_hints

doc_loader

doc_schema

formatgraph: rdflib.Graph | None = None

schemaDefs: collections.abc.MutableMapping[str, cwltool.utils.CWLObjectType]

inputs_record_schema: cwltool.utils.CWLObjectType

outputs_record_schema: cwltool.utils.CWLObjectType

container_engine = 'docker'

evalResources(builder, runtimeContext)

Parameters:

• builder (cwltool.builder.Builder)

• runtimeContext (cwltool.context.RuntimeContext)

Return type:

dict[str, Union[int, float]]

checkRequirements(rec, supported_process_requirements)

Check the presence of unsupported requirements.

Parameters:

• rec (Union[collections.abc.MutableSequence[cwltool.utils.CWLObjectType], cwltool.utils.CWLObjectType, cwltool.utils.CWLOutputType, None])

• supported_process_requirements (collections.abc.Iterable[str])

Return type:

None

validate_hints(avsc_names, hints, strict)

Process the hints field.

Parameters:

• avsc_names (schema_salad.avro.schema.Names)

• hints (list[cwltool.utils.CWLObjectType])

• strict (bool)

Return type:

None

visit(op)

Parameters:

op (Callable[[ruamel.yaml.comments.CommentedMap], None])

Return type:

None

abstract job(job_order, output_callbacks, runtimeContext)

Parameters:

• job_order (cwltool.utils.CWLObjectType)

• output_callbacks (cwltool.utils.OutputCallbackType)

• runtimeContext (cwltool.context.RuntimeContext)

Return type:

cwltool.utils.JobsGeneratorType

__str__()

Return the id of this CWL process.

Return type:

str

cwltool.process.uniquename(stem, names=None)

Construct a thread-unique name using the given stem as a prefix.

Parameters:

• stem (str)

• names (Optional[set[str]])

Return type:

str

cwltool.process.nestdir(base, deps)

Parameters:

• base (str)

• deps (cwltool.utils.CWLObjectType)

Return type:

cwltool.utils.CWLObjectType

cwltool.process.mergedirs(listing)

Parameters:

listing (collections.abc.MutableSequence[cwltool.utils.CWLObjectType])

Return type:

collections.abc.MutableSequence[cwltool.utils.CWLObjectType]

cwltool.process.CWL_IANA = 'https://www.iana.org/assignments/media-types/application/cwl'

cwltool.process.scandeps(base, doc, reffields, urlfields, loadref, urljoin=urllib.parse.urljoin, nestdirs=True)

Search for external files references in a CWL document or input object.

Looks for objects with ‘class: File’ or ‘class: Directory’ and adds them to the list of dependencies.

Parameters:

• base (str) – the base URL for relative references.

• doc (Union[cwltool.utils.CWLObjectType, collections.abc.MutableSequence[cwltool.utils.CWLObjectType]]) – a CWL document or input object

• urlfields (set[str]) – added as a File dependency

• reffields (set[str]) – field name like a workflow step ‘run’; will be added as a dependency and also loaded (using the ‘loadref’ function) and recursively scanned for dependencies. Those dependencies will be added as secondary files to the primary file.

• nestdirs (bool) – if true, create intermediate directory objects when a file is located in a subdirectory under the starting directory. This is so that if the dependencies are materialized, they will produce the same relative file system locations.

• loadref (Callable[[str, str], Union[ruamel.yaml.comments.CommentedMap, ruamel.yaml.comments.CommentedSeq, str, None]])

• urljoin (Callable[[str, str], str])

Returns:

A list of File or Directory dependencies

Return type:

collections.abc.MutableSequence[cwltool.utils.CWLObjectType]

cwltool.process.compute_checksums(fs_access, fileobj)

Parameters:

• fs_access (cwltool.stdfsaccess.StdFsAccess)

• fileobj (cwltool.utils.CWLObjectType)

Return type:

None