Warning

This document is for an in-development version of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

galaxy.tool_util package

Subpackages

Submodules

galaxy.tool_util.edam_util module

galaxy.tool_util.edam_util.load_edam_tree(path: str | None = None, *included_terms: str)[source]

galaxy.tool_util.fetcher module

class galaxy.tool_util.fetcher.ToolLocationFetcher[source]

Bases: object

__init__()[source]
to_tool_path(path_or_uri_like: StrPath, **kwds) → StrPath[source]

galaxy.tool_util.lint module

This modules contains the functions that drive the tool linting framework.

LintContext: a container for LintMessages LintMessage: the actual message and level

The idea is to define a LintContext and to apply a linting function foo on a target. The level (defined by LintLevel) determines which linting messages are shown:

lint_ctx = LintContext(level) # level is the reporting level
lint_ctx.lint(..., lint_func = foo, lint_target = target, ...)

The lint function essentially calls foo(target, self). Hence the function foo must have two parameters:

  1. the object to lint

  2. the lint context

In foo the lint context can be used to add LintMessages to the lint context by using its valid, info, warn, and error functions:

lint_foo(target, lint_ctx):
    lint_ctx.error("target is screwed")

Calling lint prints out the messages emmited by the linter function immediately. Which messages are shown can be determined with the level argument of the LintContext constructor. If set to SILENT, no messages will be printed.

For special lint targets it might be useful to have additional information in the lint messages. This can be achieved by subclassing LintMessage. See for instance XMLLintMessageLine which has an additional argument node in its constructor which is used to determine the line and filename in an XML document that caused the message.

In order to use this.

  • the lint context needs to be initialized with the additional parameter lint_message_class=XMLLintMessageLine

  • the additional parameter needs to be added as well to calls adding messages to the lint context, e.g. lint_ctx.error("some message", node=X). Note that the additional properties must be given as keyword arguments.

class galaxy.tool_util.lint.LintLevel(value)[source]

Bases: IntEnum

An enumeration.

SILENT = 5
ERROR = 4
WARN = 3
INFO = 2
VALID = 1
ALL = 0
class galaxy.tool_util.lint.Linter[source]

Bases: ABC

a linter. needs to define a lint method and the code property. optionally a fix method can be given

abstract classmethod lint(tool_source: ToolSource, lint_ctx: LintContext)[source]

should add at most one message to the lint context

classmethod name() → str[source]

get the linter name

classmethod list_linters() → List[str][source]

list the names of all linter derived from Linter

classmethod list_listers() → List[str]

list the names of all linter derived from Linter

class galaxy.tool_util.lint.LintMessage(level: str, message: str, linter: str | None = None, **kwargs)[source]

Bases: object

a message from the linter

__init__(level: str, message: str, linter: str | None = None, **kwargs)[source]
class galaxy.tool_util.lint.XMLLintMessageLine(level: str, message: str, linter: str | None = None, node: Element | None = None)[source]

Bases: LintMessage

__init__(level: str, message: str, linter: str | None = None, node: Element | None = None)[source]
class galaxy.tool_util.lint.XMLLintMessageXPath(level: str, message: str, linter: str | None = None, node: Element | None = None)[source]

Bases: LintMessage

__init__(level: str, message: str, linter: str | None = None, node: Element | None = None)[source]
class galaxy.tool_util.lint.LintContext(level: ~galaxy.tool_util.lint.LintLevel | str, lint_message_class: ~typing.Type[~galaxy.tool_util.lint.LintMessage] = <class 'galaxy.tool_util.lint.LintMessage'>, skip_types: ~typing.List[str] | None = None, object_name: str | None = None)[source]

Bases: object

__init__(level: ~galaxy.tool_util.lint.LintLevel | str, lint_message_class: ~typing.Type[~galaxy.tool_util.lint.LintMessage] = <class 'galaxy.tool_util.lint.LintMessage'>, skip_types: ~typing.List[str] | None = None, object_name: str | None = None)[source]
skip_types: List[str]
level: LintLevel
lint_message_class: Type[LintMessage]
object_name: str | None
message_list: List[LintMessage]
property found_errors: bool
property found_warns: bool
lint(name: str, lint_func: Callable[[LintTargetType, LintContext], None], lint_target: LintTargetType, module_name: str | None = None)[source]
property valid_messages: List[LintMessage]
property info_messages: List[LintMessage]
property warn_messages: List[LintMessage]
property error_messages: List[LintMessage]
valid(message: str, linter: str | None = None, *args, **kwargs) → None[source]
info(message: str, linter: str | None = None, *args, **kwargs) → None[source]
error(message: str, linter: str | None = None, *args, **kwargs) → None[source]
warn(message: str, linter: str | None = None, *args, **kwargs) → None[source]
failed(fail_level: LintLevel | str) → bool[source]
galaxy.tool_util.lint.lint_tool_source(tool_source, level=LintLevel.ALL, fail_level=LintLevel.WARN, extra_modules=None, skip_types=None, name=None) → bool[source]

apply all (applicable) linters from the linters submodule and the ones in extramodules

immediately print linter messages (wrt level) and return if linting failed (wrt fail_level)

galaxy.tool_util.lint.get_lint_context_for_tool_source(tool_source, extra_modules=None, skip_types=None, name=None) → LintContext[source]

this is the silent variant of lint_tool_source it returns the LintContext from which all linter messages and the status can be obtained

galaxy.tool_util.lint.lint_xml(tool_xml, level=LintLevel.ALL, fail_level=LintLevel.WARN, lint_message_class=<class 'galaxy.tool_util.lint.LintMessage'>, extra_modules=None, skip_types=None, name=None) → bool[source]

lint an xml tool

galaxy.tool_util.lint.lint_tool_source_with(lint_context, tool_source, extra_modules=None) → LintContext[source]
galaxy.tool_util.lint.lint_tool_source_with_modules(lint_context: LintContext, tool_source, linter_modules) → LintContext[source]
galaxy.tool_util.lint.lint_xml_with(lint_context, tool_xml, extra_modules=None) → LintContext[source]

galaxy.tool_util.loader module

galaxy.tool_util.loader.imported_macro_paths(root: Element) → List[str][source]
galaxy.tool_util.loader.load_tool(path: StrPath) → ElementTree
galaxy.tool_util.loader.load_tool_with_refereces(path: StrPath) → Tuple[ElementTree, List[str] | None]

Load XML documentation from file system and preprocesses XML macros.

Return the XML representation of the expanded tree and paths to referenced files that were imported (macros).

galaxy.tool_util.loader.raw_tool_xml_tree(path: StrPath) → ElementTree

Load raw (no macro expansion) tree representation of XML represented at the specified path.

galaxy.tool_util.loader.template_macro_params(root: Element) → Dict[str, str | None][source]

Look for template macros and populate param_dict (for cheetah) with these.

galaxy.tool_util.loader_directory module

Utilities for loading and reasoning about unparsed tools in directories.

galaxy.tool_util.loader_directory.find_possible_tools_from_path(path, recursive=False, enable_beta_formats=False)[source]

Walk a directory and find potential tool files.

galaxy.tool_util.loader_directory.is_a_yaml_with_class(path, classes)[source]

Determine if a file is a valid YAML with a supplied class entry.

galaxy.tool_util.loader_directory.is_tool_load_error(obj)[source]

Predicate to determine if object loaded for tool is a tool error.

galaxy.tool_util.loader_directory.load_tool_elements_from_path(path, load_exception_handler=<function load_exception_handler>, recursive=False, register_load_errors=False)[source]

Walk a directory and load tool XML elements.

galaxy.tool_util.loader_directory.load_tool_sources_from_path(path, load_exception_handler=<function load_exception_handler>, recursive=False, register_load_errors=False)[source]

Walk a directory and ToolSource objects.

galaxy.tool_util.loader_directory.looks_like_a_cwl_artifact(path, classes=None)[source]

Quick check to see if a file looks like it may be a CWL artifact.

galaxy.tool_util.loader_directory.looks_like_a_tool_cwl(path)[source]

Quick check to see if a file looks like it may be a CWL tool.

galaxy.tool_util.loader_directory.looks_like_a_tool_xml(path)[source]

Quick check to see if a file looks like it may be a Galaxy XML tool file.

galaxy.tool_util.loader_directory.looks_like_a_tool_yaml(path)[source]

Quick check to see if a file looks like it may be a Galaxy YAML tool file.

galaxy.tool_util.output_checker module

class galaxy.tool_util.output_checker.DETECTED_JOB_STATE(value)[source]

Bases: str, Enum

An enumeration.

OK = 'ok'
OUT_OF_MEMORY_ERROR = 'oom_error'
GENERIC_ERROR = 'generic_error'
class galaxy.tool_util.output_checker.JobMessage[source]

Bases: TypedDict

desc: str | None
code_desc: typing_extensions.NotRequired[str | None]
error_level: float
class galaxy.tool_util.output_checker.RegexJobMessage[source]

Bases: JobMessage

type: typing_extensions.Literal[regex]
stream: str | None
match: str | None
desc: str | None
code_desc: typing_extensions.NotRequired[str | None]
error_level: float
class galaxy.tool_util.output_checker.ExitCodeJobMessage[source]

Bases: JobMessage

type: typing_extensions.Literal[exit_code]
exit_code: int
desc: str | None
code_desc: typing_extensions.NotRequired[str | None]
error_level: float
class galaxy.tool_util.output_checker.MaxDiscoveredFilesJobMessage[source]

Bases: JobMessage

type: typing_extensions.Literal[max_discovered_files]
desc: str | None
code_desc: typing_extensions.NotRequired[str | None]
error_level: float
galaxy.tool_util.output_checker.check_output_regex(regex: ToolStdioRegex, stream: str, stream_name: str, job_messages: List[ExitCodeJobMessage | RegexJobMessage | MaxDiscoveredFilesJobMessage], max_error_level: int) → int[source]

check a single regex against a stream

regex the regex to check stream the stream to search in job_messages a list where the descriptions of the detected regexes can be appended max_error_level the maximum error level that has been detected so far returns the max of the error_level of the regex and the given max_error_level

galaxy.tool_util.output_checker.check_output(stdio_regexes: List[ToolStdioRegex], stdio_exit_codes: List[ToolStdioExitCode], stdout: str, stderr: str, tool_exit_code: int) → Tuple[str, str, str, List[ExitCodeJobMessage | RegexJobMessage | MaxDiscoveredFilesJobMessage]][source]

Check the output of a tool - given the stdout, stderr, and the tool’s exit code, return DETECTED_JOB_STATE.OK if the tool exited successfully or error type otherwise. No exceptions should be thrown. If this code encounters an exception, it returns OK so that the workflow can continue; otherwise, a bug in this code could halt workflow progress.

Note that, if the tool did not define any exit code handling or any stdio/stderr handling, then it reverts back to previous behavior: if stderr contains anything, then False is returned.

galaxy.tool_util.provided_metadata module

galaxy.tool_util.provided_metadata.parse_tool_provided_metadata(meta_file, provided_metadata_style=None, job_wrapper=None)[source]

Return a ToolProvidedMetadata object for specified file path.

If meta_file is absent, return a NullToolProvidedMetadata. If provided_metadata_style is None attempt to guess tool provided metadata type.

class galaxy.tool_util.provided_metadata.BaseToolProvidedMetadata[source]

Bases: object

get_new_datasets(output_name)[source]

Find new datasets for dataset discovery for specified output.

Return a list of such datasets.

Called only in the context of discovering datasets when discover_via=”tool_provided_metadata” is defined in the tool.

has_failed_outputs()[source]

Determine if generation of any of the outputs failed.

If True, this method should also log information about at least the first such failed output.

get_new_dataset_meta_by_basename(output_name, basename)[source]

For a discovered dataset, get the corresponding metadata entry.

The discovery may have been from explicit listing in this file (returned from get_new_datasets) or via file regex, either way the basename of the file is used to index the fetching of the metadata entry.

get_unnamed_outputs()[source]

Return unnamed outputs dataset introduced for upload 2.0.

Needs more formal specification but see output_collect for how destinations, types, elements, etc… are consumed.

get_dataset_meta(output_name, dataset_id, dataset_uuid)[source]

Return primary dataset metadata for specified output.

rewrite()[source]

Write metadata back to the file system.

If metadata has not changed via outputs specified as mutable, the implementation class may opt to not re-write the file.

get_new_datasets_for_metadata_collection()[source]

Return all datasets tracked that are not explicit primary outputs.

class galaxy.tool_util.provided_metadata.NullToolProvidedMetadata[source]

Bases: BaseToolProvidedMetadata

class galaxy.tool_util.provided_metadata.LegacyToolProvidedMetadata(meta_file, job_wrapper=None)[source]

Bases: BaseToolProvidedMetadata

__init__(meta_file, job_wrapper=None)[source]
get_dataset_meta(output_name, dataset_id, dataset_uuid)[source]

Return primary dataset metadata for specified output.

get_new_dataset_meta_by_basename(output_name, basename)[source]

For a discovered dataset, get the corresponding metadata entry.

The discovery may have been from explicit listing in this file (returned from get_new_datasets) or via file regex, either way the basename of the file is used to index the fetching of the metadata entry.

get_new_datasets(output_name)[source]

Find new datasets for dataset discovery for specified output.

Return a list of such datasets.

Called only in the context of discovering datasets when discover_via=”tool_provided_metadata” is defined in the tool.

has_failed_outputs()[source]

Determine if generation of any of the outputs failed.

If True, this method should also log information about at least the first such failed output.

get_unnamed_outputs()[source]

Return unnamed outputs dataset introduced for upload 2.0.

Needs more formal specification but see output_collect for how destinations, types, elements, etc… are consumed.

rewrite()[source]

Write metadata back to the file system.

If metadata has not changed via outputs specified as mutable, the implementation class may opt to not re-write the file.

get_new_datasets_for_metadata_collection()[source]

Return all datasets tracked that are not explicit primary outputs.

class galaxy.tool_util.provided_metadata.ToolProvidedMetadata(meta_file)[source]

Bases: BaseToolProvidedMetadata

__init__(meta_file)[source]
get_dataset_meta(output_name, dataset_id, dataset_uuid)[source]

Return primary dataset metadata for specified output.

get_new_dataset_meta_by_basename(output_name, basename)[source]

For a discovered dataset, get the corresponding metadata entry.

The discovery may have been from explicit listing in this file (returned from get_new_datasets) or via file regex, either way the basename of the file is used to index the fetching of the metadata entry.

get_new_datasets(output_name)[source]

Find new datasets for dataset discovery for specified output.

Return a list of such datasets.

Called only in the context of discovering datasets when discover_via=”tool_provided_metadata” is defined in the tool.

has_failed_outputs()[source]

Determine if generation of any of the outputs failed.

If True, this method should also log information about at least the first such failed output.

get_unnamed_outputs()[source]

Return unnamed outputs dataset introduced for upload 2.0.

Needs more formal specification but see output_collect for how destinations, types, elements, etc… are consumed.

rewrite()[source]

Write metadata back to the file system.

If metadata has not changed via outputs specified as mutable, the implementation class may opt to not re-write the file.