galaxy.tool_util.cwl package

galaxy.tool_util.cwl.tool_proxy(tool_path=None, tool_object=None, strict_cwl_validation=True, tool_directory=None, uuid=None) → ToolProxy[source]: Provide a proxy object to cwltool data structures to just grab relevant data.

galaxy.tool_util.cwl.tool_proxy_from_persistent_representation(persisted_tool, strict_cwl_validation=True, tool_directory=None) → ToolProxy[source]: Load a ToolProxy from a previously persisted representation.

galaxy.tool_util.cwl.workflow_proxy(workflow_path, strict_cwl_validation=True) → WorkflowProxy[source]

galaxy.tool_util.cwl.handle_outputs(job_directory=None)[source]

galaxy.tool_util.cwl.to_cwl_job(tool, param_dict, local_working_directory)[source]: tool is Galaxy’s representation of the tool and param_dict is the parameter dictionary with wrapped values.

galaxy.tool_util.cwl.to_galaxy_parameters(tool, as_dict)[source]: Tool is Galaxy’s representation of the tool and as_dict is a Galaxified representation of the input json (no paths, HDA references for instance).

galaxy.tool_util.cwl.needs_shell_quoting(string, pos=0, endpos=9223372036854775807)

Scan through string looking for a match, and return a corresponding match object instance.

Return None if no position in the string matches.

Submodules

galaxy.tool_util.cwl.cwltool_deps module

Logic for dealing with cwltool as an optional dependency.

Use this as the import interface for cwltool and just call ensure_cwltool_available() before using any of the imported functionality at runtime.

class galaxy.tool_util.cwl.cwltool_deps.CommentedMap(*args: Any, **kw: Any)[source]

Bases: ordereddict, CommentedBase

__init__(*args: Any, **kw: Any) → None[source]

update([E, ]**F) → None. Update D from dict/iterable E and F.[source]: If E is present and has a .keys() method, then does: for k in E: D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

insert(pos: Any, key: Any, value: Any, comment: Any | None = None) → None[source]: insert key value into given position, as defined by source YAML attach comment if provided

mlget(key: Any, default: Any = None, list_ok: Any = False) → Any[source]: multi-level get that expects dicts within dicts

get(key: Any, default: Any = None) → Any[source]: Return the value for key if key is in the dictionary, else default.

non_merged_items() → Any[source]

pop(k[, d]) → v, remove specified key and return the corresponding[source]: value. If key is not found, d is returned if given, otherwise KeyError is raised.

keys() → a set-like object providing a view on D's keys[source]

values() → an object providing a view on D's values[source]

items() → a set-like object providing a view on D's items[source]

property merge: Any

copy() → a shallow copy of od[source]

add_referent(cm: Any) → None[source]

add_yaml_merge(value: Any) → None[source]

update_key_value(key: Any) → None[source]

galaxy.tool_util.cwl.cwltool_deps.default_loader(fetcher_constructor: Callable[[Dict[str, str | Graph | bool], Session], Fetcher] | None = None, enable_dev: bool = False, doc_cache: bool = True) → Loader[source]

galaxy.tool_util.cwl.cwltool_deps.ensure_cwltool_available()[source]

Assert optional dependencies proxied via this module are available at runtime.

Throw an ImportError with a description of the problem if they do not exist.

galaxy.tool_util.cwl.cwltool_deps.getdefault(val: Any, default: Any) → Any[source]: Return the val using the default as backup in case the val is None.

class galaxy.tool_util.cwl.cwltool_deps.LoadingContext(kwargs: Dict[str, Any] | None = None)[source]

Bases: ContextBase

__init__(kwargs: Dict[str, Any] | None = None) → None[source]: Initialize the LoadingContext from the kwargs.

copy() → LoadingContext[source]: Return a copy of this LoadingContext.

galaxy.tool_util.cwl.cwltool_deps.needs_shell_quoting(string, pos=0, endpos=9223372036854775807)

Scan through string looking for a match, and return a corresponding match object instance.

Return None if no position in the string matches.

galaxy.tool_util.cwl.cwltool_deps.normalizeFilesDirs(job: MutableSequence[MutableMapping[str, Any]] | MutableMapping[str, Any] | DirectoryType | None) → None[source]

galaxy.tool_util.cwl.cwltool_deps.relink_initialworkdir(pathmapper: PathMapper, host_outdir: str, container_outdir: str, inplace_update: bool = False) → None[source]

galaxy.tool_util.cwl.cwltool_deps.resolve_and_validate_document(loadingContext: LoadingContext, workflowobj: CommentedMap | CommentedSeq, uri: str, preprocess_only: bool = False) → Tuple[LoadingContext, str][source]: Validate a CWL document.

class galaxy.tool_util.cwl.cwltool_deps.RuntimeContext(kwargs: Dict[str, Any] | None = None)[source]

Bases: ContextBase

outdir: str | None = None

tmpdir: str = ''

tmpdir_prefix: str = '/tmp/'

stagedir: str = ''

__init__(kwargs: Dict[str, Any] | None = None) → None[source]: Initialize the RuntimeContext from the kwargs.

tmp_outdir_prefix: str = ''

get_outdir() → str[source]: Return outdir or create one with tmp_outdir_prefix.

get_tmpdir() → str[source]: Return tmpdir or create one with tmpdir_prefix.

get_stagedir() → str[source]: Return stagedir or create one with tmpdir_prefix.

create_tmpdir() → str[source]: Create a temporary directory that respects tmpdir_prefix.

create_outdir() → str[source]: Create a temporary directory that respects tmp_outdir_prefix.

copy() → RuntimeContext[source]: Return a copy of this RuntimeContext.

class galaxy.tool_util.cwl.cwltool_deps.StdFsAccess(basedir: str)[source]

Bases: object

Local filesystem implementation.

__init__(basedir: str) → None[source]: Perform operations with respect to a base directory.

glob(pattern: str) → List[str][source]

open(fn: str, mode: str) → IO[Any][source]

exists(fn: str) → bool[source]

size(fn: str) → int[source]

isfile(fn: str) → bool[source]

isdir(fn: str) → bool[source]

listdir(fn: str) → List[str][source]

join(path: str, *paths: str) → str[source]

realpath(path: str) → str[source]

galaxy.tool_util.cwl.cwltool_deps.visit_class(rec: Any, cls: Iterable[Any], op: Callable[[...], Any]) → None[source]: Apply a function to with “class” in cls.

galaxy.tool_util.cwl.cwltool_deps.yaml_no_ts()

galaxy.tool_util.cwl.parser module

This module provides proxy objects around objects from the common workflow language reference implementation library cwltool. These proxies adapt cwltool to Galaxy features and abstract the library away from the rest of the framework.

galaxy.tool_util.cwl.parser.tool_proxy(tool_path=None, tool_object=None, strict_cwl_validation=True, tool_directory=None, uuid=None) → ToolProxy[source]: Provide a proxy object to cwltool data structures to just grab relevant data.

galaxy.tool_util.cwl.parser.load_job_proxy(job_directory, strict_cwl_validation=True) → JobProxy[source]

galaxy.tool_util.cwl.representation module

This module is responsible for converting between Galaxy’s tool input description and the CWL description for a job json.

class galaxy.tool_util.cwl.representation.INPUT_TYPE(value)[source]

Bases: str, Enum

An enumeration.

DATA = 'data'

INTEGER = 'integer'

FLOAT = 'float'

TEXT = 'text'

BOOLEAN = 'boolean'

SELECT = 'select'

FIELD = 'field'

CONDITIONAL = 'conditional'

DATA_COLLECTION = 'data_collection'

class galaxy.tool_util.cwl.representation.TypeRepresentation(name, galaxy_param_type, label, collection_type)[source]

Bases: tuple

name: str: Alias for field number 0

galaxy_param_type: Any: Alias for field number 1

label: str: Alias for field number 2

collection_type: str | None: Alias for field number 3

property uses_param

galaxy.tool_util.cwl.representation.type_representation_from_name(type_representation_name)[source]

galaxy.tool_util.cwl.representation.type_descriptions_for_field_types(field_types)[source]

galaxy.tool_util.cwl.representation.dataset_wrapper_to_file_json(inputs_dir, dataset_wrapper)[source]

galaxy.tool_util.cwl.representation.dataset_wrapper_to_directory_json(inputs_dir, dataset_wrapper)[source]

galaxy.tool_util.cwl.representation.collection_wrapper_to_array(inputs_dir, wrapped_value)[source]

galaxy.tool_util.cwl.representation.collection_wrapper_to_record(inputs_dir, wrapped_value)[source]

galaxy.tool_util.cwl.representation.to_cwl_job(tool, param_dict, local_working_directory)[source]: tool is Galaxy’s representation of the tool and param_dict is the parameter dictionary with wrapped values.

galaxy.tool_util.cwl.representation.to_galaxy_parameters(tool, as_dict)[source]: Tool is Galaxy’s representation of the tool and as_dict is a Galaxified representation of the input json (no paths, HDA references for instance).

galaxy.tool_util.cwl.representation.field_to_field_type(field)[source]

galaxy.tool_util.cwl.runnable module

Lighter-weight variant of Planemo runnable outputs.

galaxy.tool_util.cwl.runnable.get_outputs(path)[source]

class galaxy.tool_util.cwl.runnable.CwlWorkflowOutput(label)[source]

Bases: object

__init__(label)[source]

get_id()[source]

class galaxy.tool_util.cwl.runnable.ToolOutput(tool_output)[source]

Bases: object

__init__(tool_output)[source]

get_id()[source]

galaxy.tool_util.cwl.runtime_actions module

galaxy.tool_util.cwl.runtime_actions.handle_outputs(job_directory=None)[source]

galaxy.tool_util.cwl.schema module

Abstraction around cwltool and related libraries for loading a CWL artifact.

class galaxy.tool_util.cwl.schema.RawProcessReference(loading_context, process_object, uri)[source]

Bases: tuple

loading_context: LoadingContext: Alias for field number 0

process_object: CommentedMap: Alias for field number 1

uri: str: Alias for field number 2

class galaxy.tool_util.cwl.schema.ResolvedProcessDefinition(loading_context, uri, raw_process_reference)[source]

Bases: tuple

loading_context: LoadingContext: Alias for field number 0

uri: str: Alias for field number 1

raw_process_reference: RawProcessReference: Alias for field number 2

class galaxy.tool_util.cwl.schema.SchemaLoader(strict=True, validate=True)[source]

Bases: object

__init__(strict=True, validate=True)[source]

property raw_document_loader

loading_context()[source]

raw_process_reference(path, loading_context=None)[source]

raw_process_reference_for_object(process_object, uri=None, loading_context=None)[source]

process_definition(raw_process_reference: RawProcessReference) → ResolvedProcessDefinition[source]

tool(**kwds)[source]

galaxy.tool_util.cwl.util module

Client-centric CWL-related utilities.

Used to share code between the Galaxy test framework and other Galaxy CWL clients (e.g. Planemo).

galaxy.tool_util.cwl.util.set_basename_and_derived_properties(properties, basename)[source]

class galaxy.tool_util.cwl.util.OutputPropertiesType

Bases: TypedDict

class: str

location: str | None

path: str | None

listing: List[Any] | None

basename: str | None

nameroot: str | None

nameext: str | None

secondaryFiles: List[Any]

checksum: str

size: int

galaxy.tool_util.cwl.util.output_properties(path: str | None = None, content: bytes | None = None, basename=None, pseudo_location=False) → OutputPropertiesType[source]

galaxy.tool_util.cwl.util.abs_path_or_uri(path_or_uri: str, relative_to: str) → str[source]: Return the absolute path if this isn’t a URI, otherwise keep the URI the same.

galaxy.tool_util.cwl.util.abs_path(path_or_uri: str, relative_to: str) → str[source]: Return the absolute path if this is a file:// URI or a local path.

galaxy.tool_util.cwl.util.path_or_uri_to_uri(path_or_uri: str) → str[source]

galaxy.tool_util.cwl.util.galactic_job_json(job: Dict[str, Any], test_data_directory: str, upload_func: Callable[[UploadTarget], Dict[str, Any]], collection_create_func: Callable[[List[Dict[str, Any]], str], Dict[str, Any]], tool_or_workflow: typing_extensions.Literal[tool, workflow] = 'workflow') → Tuple[Dict[str, Any], List[Dict[str, Any]]][source]

Adapt a CWL job object to the Galaxy API.

CWL derived tools in Galaxy can consume a job description sort of like CWL job objects via the API but paths need to be replaced with datasets and records and arrays with collection references. This function will stage files and modify the job description to adapt to these changes for Galaxy.

class galaxy.tool_util.cwl.util.UploadTarget[source]: Bases: object

class galaxy.tool_util.cwl.util.FileLiteralTarget(contents: str, **kwargs)[source]

Bases: UploadTarget

__init__(contents: str, **kwargs) → None[source]

class galaxy.tool_util.cwl.util.FileUploadTarget(path: str | None, secondary_files: str | None = None, composite_data: List[str] | None = None, **kwargs)[source]

Bases: UploadTarget

__init__(path: str | None, secondary_files: str | None = None, composite_data: List[str] | None = None, **kwargs) → None[source]

class galaxy.tool_util.cwl.util.ObjectUploadTarget(the_object: Any)[source]

Bases: UploadTarget

__init__(the_object: Any) → None[source]

class galaxy.tool_util.cwl.util.DirectoryUploadTarget(tar_path: str)[source]

Bases: UploadTarget

__init__(tar_path: str) → None[source]

class galaxy.tool_util.cwl.util.GalaxyOutput(history_id, history_content_type, history_content_id, metadata)

Bases: tuple

history_content_id: Alias for field number 2

history_content_type: Alias for field number 1

history_id: Alias for field number 0

metadata: Alias for field number 3

galaxy.tool_util.cwl.util.tool_response_to_output(tool_response, history_id, output_name)[source]

galaxy.tool_util.cwl.util.invocation_to_output(invocation, history_id, output_name)[source]

galaxy.tool_util.cwl.util.output_to_cwl_json(galaxy_output, get_metadata, get_dataset, get_extra_files, pseudo_location=False)[source]

Convert objects in a Galaxy history into a CWL object.

Useful in running conformance tests and implementing the cwl-runner interface via Galaxy.

galaxy.tool_util.cwl.util.download_output(galaxy_output, get_metadata, get_dataset, get_extra_files, output_path)[source]

galaxy.tool_util.cwl.util.guess_artifact_type(path)[source]