cwltool.main

Entry point for cwltool.

Attributes

CWL_CONTENT_TYPES

jobloader_id_name

jobloaderctx

defaultStreamHandler

CWL_IANA

ga4gh_tool_registries

ALLUPDATES

UPDATES

DEFAULT_TMP_PREFIX

CWLObjectType

Typical raw dictionary found in lightly parsed CWL.

CWLOutputType

processes_to_kill

docker_exe

ProvOut

Exceptions

ArgumentException

Mismatched command line arguments provided.

GraphTargetMissingException

When a $graph is encountered and there is no target and no main/#main.

UnsupportedRequirement

Common base class for all non-exit exceptions.

WorkflowException

Common base class for all non-exit exceptions.

Classes

LoadingContext

Shared kwargs based initializer for RuntimeContext and LoadingContext.

RuntimeContext

Shared kwargs based initializer for RuntimeContext and LoadingContext.

ResearchObject

CWLProv Research Object.

WritableBagFile

Writes files in research object.

JobExecutor

Abstract base job executor.

MultithreadedJobExecutor

Experimental multi-threaded CWL executor.

SingleJobExecutor

Default single-threaded CWL reference executor.

MpiConfig

MutationManager

Lock manager for checking correctness of in-place update of files.

Process

Abstract CWL Process.

ProcessGenerator

Abstract CWL Process.

SecretStore

Minimal implementation of a secret storage.

DependenciesConfiguration

Dependency configuration class, for RuntimeContext.job_script_provider.

StdFsAccess

Local filesystem implementation.

HasReqsHints

Base class for get_requirement().

Workflow

Abstract CWL Process.

ProvLogFormatter

Enforce ISO8601 with both T and Z.

Functions

arg_parser()

generate_parser(toolparser, tool, namemap, records[, ...])

Generate an ArgumentParser for the given CWL Process.

get_default_args()

Get default values of cwltool's command line options.

getdefault(val, default)

Return the val using the default as backup in case the val is None.

close_ro(research_object[, save_to])

Close the Research Object, optionally saving to specified folder.

create_job(research_object, builder_job[, is_output])

Generate the new job object with RO specific relative paths.

open_log_file_for_activity(research_object, uuid_uri)

Begin the per-activity log.

packed_workflow(research_object, packed)

Pack CWL description to generate re-runnable CWL object in RO.

printdot(wf, ctx, stdout)

printrdf(wflow, ctx, style)

Serialize the CWL document into a string, ready for printing.

default_loader([fetcher_constructor, enable_dev, ...])

fetch_document(argsworkflow[, loadingContext])

Retrieve a CWL document.

load_overrides(ov, base_url)

make_tool(uri, loadingContext)

Make a Python CWL object.

resolve_and_validate_document(loadingContext, ...[, ...])

Validate a CWL document.

resolve_overrides(ov, ov_uri, baseurl)

resolve_tool_uri(argsworkflow[, resolver, ...])

configure_logging(stderr_handler, no_warnings, quiet, ...)

Configure logging.

pack(loadingContext, uri[, rewrite_out, loader])

add_sizes(fsaccess, obj)

mergedirs(listing)

scandeps(base, doc, reffields, urlfields, loadref[, ...])

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

shortname(inputid)

use_custom_schema(version, name, text)

use_standard_schema(version)

tool_resolver(document_loader, uri)

get_container_from_software_requirements(...[, ...])

get_process(tool, step_id, loading_context)

Find the underlying Process for a given Workflow step id.

get_step(tool, step_id, loading_context)

Extract a single WorkflowStep for the given step_id.

get_subgraph(roots, tool, loading_context)

Extract the subgraph for the given roots.

adjustDirObjs(rec, op)

Apply an update function to each Directory object in the object rec.

normalizeFilesDirs(job)

trim_listing(obj)

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

versionstring()

Version of CWLtool used to execute the workflow.

visit_class(rec, cls, op)

Apply a function to with "class" in cls.

append_word_to_default_user_agent(word)

Append the specified word to the requests http user agent string if it's not already there.

generate_example_input(inptype, default)

Convert a single input schema into an example.

realize_input_schema(input_types, schema_defs)

Replace references to named typed with the actual types.

generate_input_template(tool)

Generate an example input object for the given CWL process.

load_job_order(args, stdin, fetcher_constructor, ...)

init_job_order(job_order_object, args, process, ...[, ...])

make_relative(base, obj)

Relativize the location URI of a File or Directory object.

printdeps(obj, document_loader, stdout, relative_deps, uri)

Print a JSON representation of the dependencies of the CWL document.

prov_deps(obj, document_loader, uri[, basedir])

find_deps(obj, document_loader, uri[, basedir, nestdirs])

Find the dependencies of the CWL document.

print_pack(loadingContext, uri)

Return a CWL serialization of the CWL document in JSON.

supported_cwl_versions(enable_dev)

setup_schema(args, custom_schema_callback)

setup_provenance(args, runtimeContext[, argsl])

setup_loadingContext(loadingContext, runtimeContext, args)

Prepare a LoadingContext from the given arguments.

make_template(tool, target)

Make a template CWL input object for the give Process.

inherit_reqshints(tool, parent)

Copy down requirements and hints from ancestors of a given process.

choose_target(args, tool, loading_context)

Walk the Workflow, extract the subset matches all the args.targets.

choose_step(args, tool, loading_context)

Walk the given Workflow and extract just args.single_step.

choose_process(args, tool, loadingContext)

Walk the given Workflow and extract just args.single_process.

check_working_directories(runtimeContext)

Make any needed working directories.

print_targets(tool, stdout, loading_context[, prefix])

Recursively find targets for --subgraph and friends.

main([argsl, args, job_order_object, stdin, stdout, ...])

find_default_container(builder[, default_container, ...])

Find a container.

windows_check()

See if we are running on MS Windows and warn about the lack of support.

run(*args, **kwargs)

Run cwltool.

Module Contents

cwltool.main.CWL_CONTENT_TYPES = ['text/plain', 'application/json', 'text/vnd.yaml', 'text/yaml', 'text/x-yaml',...
cwltool.main.arg_parser()
Return type:

argparse.ArgumentParser

cwltool.main.generate_parser(toolparser, tool, namemap, records, input_required=True, urljoin=urllib.parse.urljoin, base_uri='')

Generate an ArgumentParser for the given CWL Process.

Parameters:
Return type:

argparse.ArgumentParser

cwltool.main.get_default_args()

Get default values of cwltool’s command line options.

Return type:

Dict[str, Any]

class cwltool.main.LoadingContext(kwargs=None)

Bases: ContextBase

Inheritance diagram of cwltool.main.LoadingContext

Shared kwargs based initializer for RuntimeContext and LoadingContext.

Parameters:

kwargs (Optional[Dict[str, Any]])

copy()

Return a copy of this LoadingContext.

Return type:

LoadingContext

class cwltool.main.RuntimeContext(kwargs=None)

Bases: ContextBase

Inheritance diagram of cwltool.main.RuntimeContext

Shared kwargs based initializer for RuntimeContext and LoadingContext.

Parameters:

kwargs (Optional[Dict[str, Any]])

outdir: str | None = None
tmpdir: str = ''
tmpdir_prefix: str
tmp_outdir_prefix: str = ''
stagedir: str = ''
get_outdir()

Return outdir or create one with tmp_outdir_prefix.

Return type:

str

get_tmpdir()

Return tmpdir or create one with tmpdir_prefix.

Return type:

str

get_stagedir()

Return stagedir or create one with tmpdir_prefix.

Return type:

str

create_tmpdir()

Create a temporary directory that respects tmpdir_prefix.

Return type:

str

create_outdir()

Create a temporary directory that respects tmp_outdir_prefix.

Return type:

str

copy()

Return a copy of this RuntimeContext.

Return type:

RuntimeContext

cwltool.main.getdefault(val, default)

Return the val using the default as backup in case the val is None.

Parameters:
  • val (Any)

  • default (Any)

Return type:

Any

class cwltool.main.ResearchObject(fsaccess, temp_prefix_ro='tmp', orcid='', full_name='')

CWLProv Research Object.

Parameters:
self_check()

Raise ValueError if this RO is closed.

Return type:

None

__str__()

Represent this RO as a string.

Return type:

str

user_provenance(document)

Add the user provenance.

Parameters:

document (prov.model.ProvDocument)

Return type:

None

add_tagfile(path, timestamp=None)

Add tag files to our research object.

Parameters:
Return type:

None

add_uri(uri, timestamp=None)
Parameters:
Return type:

cwltool.cwlprov.Aggregate

add_annotation(about, content, motivated_by='oa:describing')

Cheap URI relativize for current directory and /.

Parameters:
  • about (str)

  • content (List[str])

  • motivated_by (str)

Return type:

str

generate_snapshot(prov_dep)

Copy all of the CWL files to the snapshot/ directory.

Parameters:

prov_dep (cwltool.utils.CWLObjectType)

Return type:

None

has_data_file(sha1hash)

Confirm the presence of the given file in the RO.

Parameters:

sha1hash (str)

Return type:

bool

add_data_file(from_fp, timestamp=None, content_type=None)

Copy inputs to data/ folder.

Parameters:
Return type:

str

add_to_manifest(rel_path, checksums)

Add files to the research object manifest.

Parameters:
Return type:

None

class cwltool.main.WritableBagFile(research_object, rel_path)

Bases: io.FileIO

Inheritance diagram of cwltool.main.WritableBagFile

Writes files in research object.

Parameters:
write(b)

Write some content to the Bag.

Parameters:

b (Any)

Return type:

int

close()

Flush and close this stream.

Finalize checksums and manifests.

Return type:

None

seekable()

Return False, seeking is not supported.

Return type:

bool

readable()

Return False, reading is not supported.

Return type:

bool

truncate(size=None)

Resize the stream, only if we haven’t started writing.

Parameters:

size (Optional[int])

Return type:

int

cwltool.main.close_ro(research_object, save_to=None)

Close the Research Object, optionally saving to specified folder.

Closing will remove any temporary files used by this research object. After calling this method, this ResearchObject instance can no longer be used, except for no-op calls to .close().

The ‘saveTo’ folder should not exist - if it does, it will be deleted.

It is safe to call this function multiple times without the ‘saveTo’ argument, e.g. within a try..finally block to ensure the temporary files of this Research Object are removed.

Parameters:
Return type:

None

cwltool.main.create_job(research_object, builder_job, is_output=False)

Generate the new job object with RO specific relative paths.

Parameters:
Return type:

cwltool.utils.CWLObjectType

cwltool.main.open_log_file_for_activity(research_object, uuid_uri)

Begin the per-activity log.

Parameters:
Return type:

Union[io.TextIOWrapper, WritableBagFile]

cwltool.main.packed_workflow(research_object, packed)

Pack CWL description to generate re-runnable CWL object in RO.

Parameters:
Return type:

None

cwltool.main.printdot(wf, ctx, stdout)
Parameters:
Return type:

None

cwltool.main.printrdf(wflow, ctx, style)

Serialize the CWL document into a string, ready for printing.

Parameters:
Return type:

str

exception cwltool.main.ArgumentException

Bases: Exception

Inheritance diagram of cwltool.main.ArgumentException

Mismatched command line arguments provided.

exception cwltool.main.GraphTargetMissingException

Bases: WorkflowException

Inheritance diagram of cwltool.main.GraphTargetMissingException

When a $graph is encountered and there is no target and no main/#main.

exception cwltool.main.UnsupportedRequirement

Bases: WorkflowException

Inheritance diagram of cwltool.main.UnsupportedRequirement

Common base class for all non-exit exceptions.

exception cwltool.main.WorkflowException

Bases: Exception

Inheritance diagram of cwltool.main.WorkflowException

Common base class for all non-exit exceptions.

class cwltool.main.JobExecutor

Abstract base job executor.

__call__(process, job_order_object, runtime_context, logger=_logger)
Parameters:
Return type:

Tuple[Optional[cwltool.utils.CWLObjectType], str]

output_callback(out, process_status)

Collect the final status and outputs.

Parameters:
  • out (Optional[cwltool.utils.CWLObjectType])

  • process_status (str)

Return type:

None

abstract run_jobs(process, job_order_object, logger, runtime_context)

Execute the jobs for the given Process.

Parameters:
Return type:

None

execute(process, job_order_object, runtime_context, logger=_logger)

Execute the process.

Parameters:
Return type:

Tuple[Union[Optional[cwltool.utils.CWLObjectType]], str]

class cwltool.main.MultithreadedJobExecutor

Bases: JobExecutor

Inheritance diagram of cwltool.main.MultithreadedJobExecutor

Experimental multi-threaded CWL executor.

Does simple resource accounting, will not start a job unless it has cores / ram available, but does not make any attempt to optimize usage.

select_resources(request, runtime_context)

Naïve check for available cpu cores and memory.

Parameters:
Return type:

Dict[str, Union[int, float]]

run_job(job, runtime_context)

Execute a single Job in a separate thread.

Parameters:
Return type:

None

wait_for_next_completion(runtime_context)

Wait for jobs to finish.

Parameters:

runtime_context (cwltool.context.RuntimeContext)

Return type:

None

run_jobs(process, job_order_object, logger, runtime_context)

Execute the jobs for the given Process.

Parameters:
Return type:

None

class cwltool.main.SingleJobExecutor

Bases: JobExecutor

Inheritance diagram of cwltool.main.SingleJobExecutor

Default single-threaded CWL reference executor.

run_jobs(process, job_order_object, logger, runtime_context)

Execute the jobs for the given Process.

Parameters:
Return type:

None

cwltool.main.default_loader(fetcher_constructor=None, enable_dev=False, doc_cache=True)
Parameters:
  • fetcher_constructor (Optional[schema_salad.utils.FetcherCallableType])

  • enable_dev (bool)

  • doc_cache (bool)

Return type:

schema_salad.ref_resolver.Loader

cwltool.main.fetch_document(argsworkflow, loadingContext=None)

Retrieve a CWL document.

Parameters:
Return type:

Tuple[cwltool.context.LoadingContext, ruamel.yaml.comments.CommentedMap, str]

cwltool.main.jobloader_id_name = '__id'
cwltool.main.jobloaderctx: schema_salad.utils.ContextType
cwltool.main.load_overrides(ov, base_url)
Parameters:
Return type:

List[cwltool.utils.CWLObjectType]

cwltool.main.make_tool(uri, loadingContext)

Make a Python CWL object.

Parameters:
Return type:

cwltool.process.Process

cwltool.main.resolve_and_validate_document(loadingContext, workflowobj, uri, preprocess_only=False)

Validate a CWL document.

Parameters:
Return type:

Tuple[cwltool.context.LoadingContext, str]

cwltool.main.resolve_overrides(ov, ov_uri, baseurl)
Parameters:
  • ov (schema_salad.utils.IdxResultType)

  • ov_uri (str)

  • baseurl (str)

Return type:

List[cwltool.utils.CWLObjectType]

cwltool.main.resolve_tool_uri(argsworkflow, resolver=None, fetcher_constructor=None, document_loader=None)
Parameters:
  • argsworkflow (str)

  • resolver (Optional[cwltool.utils.ResolverType])

  • fetcher_constructor (Optional[schema_salad.utils.FetcherCallableType])

  • document_loader (Optional[schema_salad.ref_resolver.Loader])

Return type:

Tuple[str, str]

cwltool.main.configure_logging(stderr_handler, no_warnings, quiet, debug, enable_color, timestamps, base_logger=_logger)

Configure logging.

Parameters:
Return type:

None

cwltool.main.defaultStreamHandler
class cwltool.main.MpiConfig(runner='mpirun', nproc_flag='-n', default_nproc=1, extra_flags=None, env_pass=None, env_pass_regex=None, env_set=None)
Parameters:
  • runner (str)

  • nproc_flag (str)

  • default_nproc (Union[int, str])

  • extra_flags (Optional[List[str]])

  • env_pass (Optional[List[str]])

  • env_pass_regex (Optional[List[str]])

  • env_set (Optional[Mapping[str, str]])

classmethod load(config_file_name)

Create the MpiConfig object from the contents of a YAML file.

The file must contain exactly one object, whose attributes must be in the list allowed in the class initialiser (all are optional).

Parameters:

config_file_name (str)

Return type:

MpiConfigT

pass_through_env_vars(env)

Take the configured list of environment variables and pass them to the executed process.

Parameters:

env (MutableMapping[str, str])

Return type:

None

set_env_vars(env)

Set some variables to the value configured.

Parameters:

env (MutableMapping[str, str])

Return type:

None

class cwltool.main.MutationManager

Lock manager for checking correctness of in-place update of files.

Used to validate that in-place file updates happen sequentially, and that a file which is registered for in-place update cannot be read or updated by any other steps.

register_reader(stepname, obj)
Parameters:
  • stepname (str)

  • obj (cwltool.utils.CWLObjectType)

Return type:

None

release_reader(stepname, obj)
Parameters:
  • stepname (str)

  • obj (cwltool.utils.CWLObjectType)

Return type:

None

register_mutation(stepname, obj)
Parameters:
  • stepname (str)

  • obj (cwltool.utils.CWLObjectType)

Return type:

None

set_generation(obj)
Parameters:

obj (cwltool.utils.CWLObjectType)

Return type:

None

unset_generation(obj)
Parameters:

obj (cwltool.utils.CWLObjectType)

Return type:

None

cwltool.main.pack(loadingContext, uri, rewrite_out=None, loader=None)
Parameters:
Return type:

cwltool.utils.CWLObjectType

cwltool.main.CWL_IANA = 'https://www.iana.org/assignments/media-types/application/cwl'
class cwltool.main.Process(toolpath_object, loadingContext)

Bases: cwltool.utils.HasReqsHints

Inheritance diagram of cwltool.main.Process

Abstract CWL Process.

Parameters:
evalResources(builder, runtimeContext)
Parameters:
Return type:

Dict[str, Union[int, float]]

checkRequirements(rec, supported_process_requirements)

Check the presence of unsupported requirements.

Parameters:
  • rec (Union[MutableSequence[cwltool.utils.CWLObjectType], cwltool.utils.CWLObjectType, cwltool.utils.CWLOutputType, None])

  • supported_process_requirements (Iterable[str])

Return type:

None

validate_hints(avsc_names, hints, strict)

Process the hints field.

Parameters:
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 (Optional[cwltool.utils.OutputCallbackType])

  • runtimeContext (cwltool.context.RuntimeContext)

Return type:

cwltool.utils.JobsGeneratorType

__str__()

Return the id of this CWL process.

Return type:

str

cwltool.main.add_sizes(fsaccess, obj)
Parameters:
Return type:

None

cwltool.main.mergedirs(listing)
Parameters:

listing (MutableSequence[cwltool.utils.CWLObjectType])

Return type:

MutableSequence[cwltool.utils.CWLObjectType]

cwltool.main.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, 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:

MutableSequence[cwltool.utils.CWLObjectType]

cwltool.main.shortname(inputid)
Parameters:

inputid (str)

Return type:

str

cwltool.main.use_custom_schema(version, name, text)
Parameters:
Return type:

None

cwltool.main.use_standard_schema(version)
Parameters:

version (str)

Return type:

None

class cwltool.main.ProcessGenerator(toolpath_object, loadingContext)

Bases: cwltool.process.Process

Inheritance diagram of cwltool.main.ProcessGenerator

Abstract CWL Process.

Parameters:
job(job_order, output_callbacks, runtimeContext)
Parameters:
  • job_order (cwltool.utils.CWLObjectType)

  • output_callbacks (Optional[cwltool.utils.OutputCallbackType])

  • runtimeContext (cwltool.context.RuntimeContext)

Return type:

cwltool.utils.JobsGeneratorType

result(job_order, jobout, runtimeContext)
Parameters:
Return type:

Tuple[cwltool.process.Process, cwltool.utils.CWLObjectType]

cwltool.main.ga4gh_tool_registries = ['https://dockstore.org/api']
cwltool.main.tool_resolver(document_loader, uri)
Parameters:
Return type:

Optional[str]

class cwltool.main.SecretStore

Minimal implementation of a secret storage.

add(value)

Add the given value to the store.

Returns a placeholder value to use until the real value is needed.

Parameters:

value (Optional[cwltool.utils.CWLOutputType])

Return type:

Optional[cwltool.utils.CWLOutputType]

store(secrets, job)

Sanitize the job object of any of the given secrets.

Parameters:
  • secrets (List[str])

  • job (cwltool.utils.CWLObjectType)

Return type:

None

has_secret(value)

Test if the provided document has any of our secrets.

Parameters:

value (cwltool.utils.CWLOutputType)

Return type:

bool

retrieve(value)

Replace placeholders with their corresponding secrets.

Parameters:

value (cwltool.utils.CWLOutputType)

Return type:

cwltool.utils.CWLOutputType

class cwltool.main.DependenciesConfiguration(args)

Dependency configuration class, for RuntimeContext.job_script_provider.

Parameters:

args (argparse.Namespace)

build_job_script(builder, command)
Parameters:
Return type:

str

cwltool.main.get_container_from_software_requirements(use_biocontainers, builder, container_image_cache_path='.')
Parameters:
  • use_biocontainers (bool)

  • builder (cwltool.utils.HasReqsHints)

  • container_image_cache_path (Optional[str])

Return type:

Optional[str]

class cwltool.main.StdFsAccess(basedir)

Local filesystem implementation.

Parameters:

basedir (str)

glob(pattern)
Parameters:

pattern (str)

Return type:

List[str]

open(fn, mode)
Parameters:
Return type:

IO[Any]

exists(fn)
Parameters:

fn (str)

Return type:

bool

size(fn)
Parameters:

fn (str)

Return type:

int

isfile(fn)
Parameters:

fn (str)

Return type:

bool

isdir(fn)
Parameters:

fn (str)

Return type:

bool

listdir(fn)
Parameters:

fn (str)

Return type:

List[str]

join(path, *paths)
Parameters:
Return type:

str

realpath(path)
Parameters:

path (str)

Return type:

str

cwltool.main.get_process(tool, step_id, loading_context)

Find the underlying Process for a given Workflow step id.

Parameters:
Return type:

Tuple[Any, cwltool.workflow.WorkflowStep]

cwltool.main.get_step(tool, step_id, loading_context)

Extract a single WorkflowStep for the given step_id.

Parameters:
Return type:

ruamel.yaml.comments.CommentedMap

cwltool.main.get_subgraph(roots, tool, loading_context)

Extract the subgraph for the given roots.

Parameters:
Return type:

ruamel.yaml.comments.CommentedMap

cwltool.main.ALLUPDATES
cwltool.main.UPDATES: Dict[str, Callable[[ruamel.yaml.comments.CommentedMap, schema_salad.ref_resolver.Loader, str], Tuple[ruamel.yaml.comments.CommentedMap, str]] | None]
cwltool.main.DEFAULT_TMP_PREFIX
cwltool.main.CWLObjectType

Typical raw dictionary found in lightly parsed CWL.

cwltool.main.CWLOutputType
class cwltool.main.HasReqsHints

Base class for get_requirement().

get_requirement(feature)

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

Parameters:

feature (str)

Return type:

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

cwltool.main.adjustDirObjs(rec, op)

Apply an update function to each Directory object in the object rec.

Parameters:
  • rec (Any)

  • op (Union[Callable[Ellipsis, Any], partial[Any]])

Return type:

None

cwltool.main.normalizeFilesDirs(job)
Parameters:

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

Return type:

None

cwltool.main.processes_to_kill: Deque[subprocess.Popen[str]]
cwltool.main.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.main.versionstring()

Version of CWLtool used to execute the workflow.

Return type:

str

cwltool.main.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

class cwltool.main.Workflow(toolpath_object, loadingContext)

Bases: cwltool.process.Process

Inheritance diagram of cwltool.main.Workflow

Abstract CWL Process.

Parameters:
make_workflow_step(toolpath_object, pos, loadingContext, parentworkflowProv=None)
Parameters:
Return type:

WorkflowStep

job(job_order, output_callbacks, runtimeContext)
Parameters:
  • job_order (cwltool.utils.CWLObjectType)

  • output_callbacks (Optional[cwltool.utils.OutputCallbackType])

  • runtimeContext (cwltool.context.RuntimeContext)

Return type:

cwltool.utils.JobsGeneratorType

visit(op)
Parameters:

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

Return type:

None

cwltool.main.docker_exe: str
cwltool.main.append_word_to_default_user_agent(word)

Append the specified word to the requests http user agent string if it’s not already there.

Parameters:

word (str)

Return type:

None

cwltool.main.generate_example_input(inptype, default)

Convert a single input schema into an example.

Parameters:
  • inptype (Optional[cwltool.utils.CWLOutputType])

  • default (Optional[cwltool.utils.CWLOutputType])

Return type:

Tuple[Any, str]

cwltool.main.realize_input_schema(input_types, schema_defs)

Replace references to named typed with the actual types.

Parameters:
  • input_types (MutableSequence[Union[str, cwltool.utils.CWLObjectType]])

  • schema_defs (MutableMapping[str, cwltool.utils.CWLObjectType])

Return type:

MutableSequence[Union[str, cwltool.utils.CWLObjectType]]

cwltool.main.generate_input_template(tool)

Generate an example input object for the given CWL process.

Parameters:

tool (cwltool.process.Process)

Return type:

cwltool.utils.CWLObjectType

cwltool.main.load_job_order(args, stdin, fetcher_constructor, overrides_list, tool_file_uri)
Parameters:
  • args (argparse.Namespace)

  • stdin (IO[Any])

  • fetcher_constructor (Optional[schema_salad.utils.FetcherCallableType])

  • overrides_list (List[cwltool.utils.CWLObjectType])

  • tool_file_uri (str)

Return type:

Tuple[Optional[cwltool.utils.CWLObjectType], str, schema_salad.ref_resolver.Loader]

cwltool.main.init_job_order(job_order_object, args, process, loader, stdout, print_input_deps=False, relative_deps='primary', make_fs_access=StdFsAccess, input_basedir='', secret_store=None, input_required=True, runtime_context=None)
Parameters:
Return type:

cwltool.utils.CWLObjectType

cwltool.main.make_relative(base, obj)

Relativize the location URI of a File or Directory object.

Parameters:
  • base (str)

  • obj (cwltool.utils.CWLObjectType)

Return type:

None

cwltool.main.printdeps(obj, document_loader, stdout, relative_deps, uri, basedir=None, nestdirs=True)

Print a JSON representation of the dependencies of the CWL document.

Parameters:
Return type:

None

cwltool.main.prov_deps(obj, document_loader, uri, basedir=None)
Parameters:
Return type:

cwltool.utils.CWLObjectType

cwltool.main.find_deps(obj, document_loader, uri, basedir=None, nestdirs=True)

Find the dependencies of the CWL document.

Parameters:
Return type:

cwltool.utils.CWLObjectType

cwltool.main.print_pack(loadingContext, uri)

Return a CWL serialization of the CWL document in JSON.

Parameters:
Return type:

str

cwltool.main.supported_cwl_versions(enable_dev)
Parameters:

enable_dev (bool)

Return type:

List[str]

cwltool.main.setup_schema(args, custom_schema_callback)
Parameters:
Return type:

None

class cwltool.main.ProvLogFormatter

Bases: logging.Formatter

Inheritance diagram of cwltool.main.ProvLogFormatter

Enforce ISO8601 with both T and Z.

formatTime(record, datefmt=None)

Override the default formatTime to include the timezone.

Parameters:
Return type:

str

cwltool.main.ProvOut
cwltool.main.setup_provenance(args, runtimeContext, argsl=None)
Parameters:
Return type:

Tuple[ProvOut, logging.StreamHandler[ProvOut]]

cwltool.main.setup_loadingContext(loadingContext, runtimeContext, args)

Prepare a LoadingContext from the given arguments.

Parameters:
Return type:

cwltool.context.LoadingContext

cwltool.main.make_template(tool, target)

Make a template CWL input object for the give Process.

Parameters:
Return type:

None

cwltool.main.inherit_reqshints(tool, parent)

Copy down requirements and hints from ancestors of a given process.

Parameters:
Return type:

None

cwltool.main.choose_target(args, tool, loading_context)

Walk the Workflow, extract the subset matches all the args.targets.

Parameters:
Return type:

Optional[cwltool.process.Process]

cwltool.main.choose_step(args, tool, loading_context)

Walk the given Workflow and extract just args.single_step.

Parameters:
Return type:

Optional[cwltool.process.Process]

cwltool.main.choose_process(args, tool, loadingContext)

Walk the given Workflow and extract just args.single_process.

Parameters:
Return type:

Optional[cwltool.process.Process]

cwltool.main.check_working_directories(runtimeContext)

Make any needed working directories.

Parameters:

runtimeContext (cwltool.context.RuntimeContext)

Return type:

Optional[int]

cwltool.main.print_targets(tool, stdout, loading_context, prefix='')

Recursively find targets for –subgraph and friends.

Parameters:
Return type:

None

cwltool.main.main(argsl=None, args=None, job_order_object=None, stdin=sys.stdin, stdout=None, stderr=sys.stderr, versionfunc=versionstring, logger_handler=None, custom_schema_callback=None, executor=None, loadingContext=None, runtimeContext=None, input_required=True)
Parameters:
Return type:

int

cwltool.main.find_default_container(builder, default_container=None, use_biocontainers=None, container_image_cache_path=None)

Find a container.

Parameters:
  • builder (cwltool.utils.HasReqsHints)

  • default_container (Optional[str])

  • use_biocontainers (Optional[bool])

  • container_image_cache_path (Optional[str])

Return type:

Optional[str]

cwltool.main.windows_check()

See if we are running on MS Windows and warn about the lack of support.

Return type:

None

cwltool.main.run(*args, **kwargs)

Run cwltool.

Parameters:
  • args (Any)

  • kwargs (Any)

Return type:

int