galaxy.util package

Utility functions used systemwide.

class galaxy.util.Element(tag, attrib={}, **extra)

Bases: _Element

find(self, path, namespaces=None)

Finds the first matching subelement, by tag name or path.

The optional namespaces argument accepts a prefix-to-namespace mapping that allows the usage of XPath prefixes in the path expression.

findall(self, path, namespaces=None)

Finds all matching subelements, by tag name or path.

The optional namespaces argument accepts a prefix-to-namespace mapping that allows the usage of XPath prefixes in the path expression.

galaxy.util.SubElement(parent: Element, tag: str, attrib: Dict[str, str] | None = None, **extra) → Element

class galaxy.util.ElementTree(element=None, file=None)

Bases: _ElementTree

getroot(self)

Gets the root element for this tree.

galaxy.util.XML(text: str | bytes) → Element

galaxy.util.str_removeprefix(s: str, prefix: str)

str.removeprefix() equivalent for Python < 3.9

galaxy.util.remove_protocol_from_url(url)

Supplied URL may be null, if not ensure http:// or https:// etc… is stripped off.

galaxy.util.is_binary(value)

File is binary if it contains a null-byte by default (e.g. behavior of grep, etc.). This may fail for utf-16 files, but so would ASCII encoding. >>> is_binary( string.printable ) False >>> is_binary( b’xcex94’ ) False >>> is_binary( b’x00’ ) True

galaxy.util.is_uuid(value)

This method returns True if value is a UUID, otherwise False. >>> is_uuid( “123e4567-e89b-12d3-a456-426655440000” ) True >>> is_uuid( “0x3242340298902834” ) False

galaxy.util.directory_hash_id(id)

>>> directory_hash_id( 100 )
['000']
>>> directory_hash_id( "90000" )
['090']
>>> directory_hash_id("777777777")
['000', '777', '777']
>>> directory_hash_id("135ee48a-4f51-470c-ae2f-ce8bd78799e6")
['1', '3', '5']

galaxy.util.get_charset_from_http_headers(headers, default=None)

galaxy.util.synchronized(func)

This wrapper will serialize access to ‘func’ to a single thread. Use it as a decorator.

galaxy.util.iter_start_of_line(fh, chunk_size=None)

Iterate over fh and call readline(chunk_size).

galaxy.util.file_reader(fp, chunk_size=65536)

This generator yields the open file object in chunks (default 64k).

galaxy.util.chunk_iterable(it: Iterable, size: int = 1000) → Iterator[tuple]

Break an iterable into chunks of size elements.

>>> list(chunk_iterable([1, 2, 3, 4, 5, 6, 7], 3))
[(1, 2, 3), (4, 5, 6), (7,)]

galaxy.util.unique_id(KEY_SIZE=128)

Generates an unique id

>>> ids = [ unique_id() for i in range(1000) ]
>>> len(set(ids))
1000

galaxy.util.parse_xml(fname: str | PathLike, strip_whitespace=True, remove_comments=True, schemafname: str | PathLike | None = None) → ElementTree

Returns a parsed xml tree

galaxy.util.parse_xml_string(xml_string: str, strip_whitespace: bool = True) → Element

galaxy.util.parse_xml_string_to_etree(xml_string: str, strip_whitespace: bool = True) → ElementTree

galaxy.util.xml_to_string(elem: Element | None, pretty: bool = False) → str

Returns a string from an xml tree.

galaxy.util.xml_element_compare(elem1, elem2)

galaxy.util.xml_element_list_compare(elem_list1, elem_list2)

galaxy.util.xml_element_to_dict(elem)

galaxy.util.pretty_print_xml(elem, level=0)

galaxy.util.get_file_size(value, default=None)

galaxy.util.shrink_stream_by_size(value, size, join_by=b'..', left_larger=True, beginning_on_size_error=False, end_on_size_error=False)

Shrinks bytes read from value to size.

value needs to implement tell/seek, so files need to be opened in binary mode. Returns unicode text with invalid characters replaced.

galaxy.util.shrink_and_unicodify(stream)

galaxy.util.shrink_string_by_size(value, size, join_by='..', left_larger=True, beginning_on_size_error=False, end_on_size_error=False)

galaxy.util.pretty_print_time_interval(time=False, precise=False, utc=False)

Get a datetime object or a int() Epoch timestamp and return a pretty string like ‘an hour ago’, ‘Yesterday’, ‘3 months ago’, ‘just now’, etc credit: http://stackoverflow.com/questions/1551382/user-friendly-time-format-in-python

galaxy.util.restore_text(text, character_map={'\t': '__tc__', '\n': '__cn__', '\r': '__cr__', '"': '__dq__', '#': '__pd__', "'": '__sq__', '<': '__lt__', '>': '__gt__', '@': '__at__', '[': '__ob__', ']': '__cb__', '{': '__oc__', '}': '__cc__'})

Restores sanitized text

galaxy.util.sanitize_text(text, valid_characters={' ', '!', '(', ')', '*', '+', ',', '-', '.', '/', '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', ':', '=', '?', 'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L', 'M', 'N', 'O', 'P', 'Q', 'R', 'S', 'T', 'U', 'V', 'W', 'X', 'Y', 'Z', '^', '_', 'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm', 'n', 'o', 'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x', 'y', 'z'}, character_map={'\t': '__tc__', '\n': '__cn__', '\r': '__cr__', '"': '__dq__', '#': '__pd__', "'": '__sq__', '<': '__lt__', '>': '__gt__', '@': '__at__', '[': '__ob__', ']': '__cb__', '{': '__oc__', '}': '__cc__'}, invalid_character='X')

Restricts the characters that are allowed in text; accepts both strings and lists of strings; non-string entities will be cast to strings.

galaxy.util.sanitize_lists_to_string(values, valid_characters={' ', '!', '(', ')', '*', '+', ',', '-', '.', '/', '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', ':', '=', '?', 'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L', 'M', 'N', 'O', 'P', 'Q', 'R', 'S', 'T', 'U', 'V', 'W', 'X', 'Y', 'Z', '^', '_', 'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm', 'n', 'o', 'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x', 'y', 'z'}, character_map={'\t': '__tc__', '\n': '__cn__', '\r': '__cr__', '"': '__dq__', '#': '__pd__', "'": '__sq__', '<': '__lt__', '>': '__gt__', '@': '__at__', '[': '__ob__', ']': '__cb__', '{': '__oc__', '}': '__cc__'}, invalid_character='X')

galaxy.util.sanitize_param(value, valid_characters={' ', '!', '(', ')', '*', '+', ',', '-', '.', '/', '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', ':', '=', '?', 'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L', 'M', 'N', 'O', 'P', 'Q', 'R', 'S', 'T', 'U', 'V', 'W', 'X', 'Y', 'Z', '^', '_', 'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm', 'n', 'o', 'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x', 'y', 'z'}, character_map={'\t': '__tc__', '\n': '__cn__', '\r': '__cr__', '"': '__dq__', '#': '__pd__', "'": '__sq__', '<': '__lt__', '>': '__gt__', '@': '__at__', '[': '__ob__', ']': '__cb__', '{': '__oc__', '}': '__cc__'}, invalid_character='X')

Clean incoming parameters (strings or lists)

galaxy.util.sanitize_for_filename(text, default=None)

Restricts the characters that are allowed in a filename portion; Returns default value or a unique id string if result is not a valid name. Method is overly aggressive to minimize possible complications, but a maximum length is not considered.

galaxy.util.find_instance_nested(item, instances)

Recursively find instances from lists, dicts, tuples.

instances should be a tuple of valid instances. Returns a dictionary, where keys are the deepest key at which an instance has been found, and the value is the matched instance.

galaxy.util.mask_password_from_url(url)

Masks out passwords from connection urls like the database connection in galaxy.ini

>>> mask_password_from_url( 'sqlite+postgresql://user:password@localhost/' )
'sqlite+postgresql://user:********@localhost/'
>>> mask_password_from_url( 'amqp://user:amqp@localhost' )
'amqp://user:********@localhost'
>>> mask_password_from_url( 'amqp://localhost')
'amqp://localhost'

galaxy.util.ready_name_for_url(raw_name)

General method to convert a string (i.e. object name) to a URL-ready slug.

>>> ready_name_for_url( "My Cool Object" )
'My-Cool-Object'
>>> ready_name_for_url( "!My Cool Object!" )
'My-Cool-Object'
>>> ready_name_for_url( "Hello₩◎ґʟⅾ" )
'Hello'

galaxy.util.which(file: str) → str | None

galaxy.util.in_directory(file, directory, local_path_module=<module 'posixpath' from '/opt/hostedtoolcache/Python/3.8.18/x64/lib/python3.8/posixpath.py'>)

Return true, if the common prefix of both is equal to directory e.g. /a/b/c/d.rst and directory is /a/b, the common prefix is /a/b. This function isn’t used exclusively for security checks, but if it is used for such checks it is assumed that directory is a “trusted” path - supplied by Galaxy or by the admin and file is something generated by a tool, configuration, external web server, or user supplied input.

local_path_module is used by Pulsar to check Windows paths while running on a POSIX-like system.

galaxy.util.merge_sorted_iterables(operator, *iterables)

>>> operator = lambda x: x
>>> list( merge_sorted_iterables( operator, [1,2,3], [4,5] ) )
[1, 2, 3, 4, 5]
>>> list( merge_sorted_iterables( operator, [4, 5], [1,2,3] ) )
[1, 2, 3, 4, 5]
>>> list( merge_sorted_iterables( operator, [1, 4, 5], [2], [3] ) )
[1, 2, 3, 4, 5]

class galaxy.util.Params(params, sanitize=True)

Bases: object

Stores and ‘sanitizes’ parameters. Alphanumeric characters and the non-alphanumeric ones that are deemed safe are let to pass through (see L{valid_chars}). Some non-safe characters are escaped to safe forms for example C{>} becomes C{__lt__} (see L{mapped_chars}). All other characters are replaced with C{X}.

Operates on string or list values only (HTTP parameters).

>>> values = { 'status':'on', 'symbols':[  'alpha', '<>', '$rm&#!' ]  }
>>> par = Params(values)
>>> par.status
'on'
>>> par.value == None      # missing attributes return None
True
>>> par.get('price', 0)
0
>>> par.symbols            # replaces unknown symbols with X
['alpha', '__lt____gt__', 'XrmX__pd__!']
>>> sorted(par.flatten())  # flattening to a list
[('status', 'on'), ('symbols', 'XrmX__pd__!'), ('symbols', '__lt____gt__'), ('symbols', 'alpha')]

NEVER_SANITIZE = ['file_data', 'url_paste', 'URL', 'filesystem_paths']

__init__(params, sanitize=True)

flatten()

Creates a tuple list from a dict with a tuple/value pair for every value that is a list

get(key, default)

update(values)

galaxy.util.xml_text(root, name=None)

Returns the text inside an element

galaxy.util.parse_resource_parameters(resource_param_file)

Code shared between jobs and workflows for reading resource parameter configuration files.

TODO: Allow YAML in addition to XML.

galaxy.util.asbool(obj)

galaxy.util.string_as_bool(string: Any) → bool

galaxy.util.string_as_bool_or_none(string)

Returns True, None or False based on the argument:

True if passed True, ‘True’, ‘Yes’, or ‘On’ None if passed None or ‘None’ False otherwise

Note: string comparison is case-insensitive so lowecase versions of those function equivalently.

galaxy.util.listify(item: None | typing_extensions.Literal[False], do_strip: bool = False) → List

galaxy.util.listify(item: str, do_strip: bool = False) → List[str]

galaxy.util.listify(item: List[ItemType] | Tuple[ItemType, ...], do_strip: bool = False) → List[ItemType]

galaxy.util.listify(item: Any, do_strip: bool = False) → List

Make a single item a single item list.

If item is a string, it is split on comma (,) characters to produce the list. Optionally, if do_strip is true, any extra whitespace around the split items is stripped.

If item is a list it is returned unchanged. If item is a tuple, it is converted to a list and returned. If item evaluates to False, an empty list is returned.

Parameters:

• item (object) – object to make a list from

• do_strip (bool) – strip whitespaces from around split items, if set to True

Return type:

list

Returns:

The input as a list

galaxy.util.commaify(amount)

galaxy.util.unicodify(value: typing_extensions.Literal[None], encoding: str = DEFAULT_ENCODING, error: str = 'replace', strip_null: bool = False, log_exception: bool = True) → None

galaxy.util.unicodify(value: Any, encoding: str = DEFAULT_ENCODING, error: str = 'replace', strip_null: bool = False, log_exception: bool = True) → str

Returns a Unicode string or None.

>>> assert unicodify(None) is None
>>> assert unicodify('simple string') == 'simple string'
>>> assert unicodify(3) == '3'
>>> assert unicodify(bytearray([115, 116, 114, 196, 169, 195, 177, 103])) == 'strĩñg'
>>> assert unicodify(Exception('strĩñg')) == 'strĩñg'
>>> assert unicodify('cómplǐcḁtëd strĩñg') == 'cómplǐcḁtëd strĩñg'
>>> s = 'cómplǐcḁtëd strĩñg'; assert unicodify(s) == s
>>> s = 'lâtín strìñg'; assert unicodify(s.encode('latin-1'), 'latin-1') == s
>>> s = 'lâtín strìñg'; assert unicodify(s.encode('latin-1')) == 'l�t�n str��g'
>>> s = 'lâtín strìñg'; assert unicodify(s.encode('latin-1'), error='ignore') == 'ltn strg'

galaxy.util.filesystem_safe_string(s, max_len=255, truncation_chars='..', strip_leading_dot=True, invalid_chars=('/',), replacement_char='_')

Strip unicode null chars, truncate at 255 characters. Optionally replace additional invalid_chars with replacement_char .

Defaults are probably only safe on linux / osx. Needs further escaping if used in shell commands

galaxy.util.smart_str(s, encoding='utf-8', strings_only=False, errors='strict')

Returns a bytestring version of ‘s’, encoded as specified in ‘encoding’.

If strings_only is True, don’t convert (some) non-string-like objects.

Adapted from an older, simpler version of django.utils.encoding.smart_str.

>>> assert smart_str(None) == b'None'
>>> assert smart_str(None, strings_only=True) is None
>>> assert smart_str(3) == b'3'
>>> assert smart_str(3, strings_only=True) == 3
>>> s = b'a bytes string'; assert smart_str(s) == s
>>> s = bytearray(b'a bytes string'); assert smart_str(s) == s
>>> assert smart_str('a simple unicode string') == b'a simple unicode string'
>>> assert smart_str('à strange ünicode ڃtring') == b'\xc3\xa0 strange \xc3\xbcnicode \xda\x83tring'
>>> assert smart_str(b'\xc3\xa0n \xc3\xabncoded utf-8 string', encoding='latin-1') == b'\xe0n \xebncoded utf-8 string'
>>> assert smart_str(bytearray(b'\xc3\xa0n \xc3\xabncoded utf-8 string'), encoding='latin-1') == b'\xe0n \xebncoded utf-8 string'

galaxy.util.strip_control_characters(s)

Strip unicode control characters from a string.

galaxy.util.object_to_string(obj)

galaxy.util.string_to_object(s)

galaxy.util.clean_multiline_string(multiline_string, sep='\n')

Dedent, split, remove first and last empty lines, rejoin.

class galaxy.util.ParamsWithSpecs(specs=None, params=None)

Bases: defaultdict

__init__(specs=None, params=None)

galaxy.util.compare_urls(url1, url2, compare_scheme=True, compare_hostname=True, compare_path=True)

galaxy.util.read_build_sites(filename, check_builds=True)

read db names to ucsc mappings from file, this file should probably be merged with the one above

galaxy.util.relativize_symlinks(path, start=None, followlinks=False)

galaxy.util.stringify_dictionary_keys(in_dict)

galaxy.util.mkstemp_ln(src, prefix='mkstemp_ln_')

From tempfile._mkstemp_inner, generate a hard link in the same dir with a random name. Created so we can persist the underlying file of a NamedTemporaryFile upon its closure.

galaxy.util.umask_fix_perms(path, umask, unmasked_perms, gid=None)

umask-friendly permissions fixing

galaxy.util.docstring_trim(docstring)

Trimming python doc strings. Taken from: http://www.python.org/dev/peps/pep-0257/

galaxy.util.metric_prefix(number: int | float, base: int) → Tuple[float, str]

>>> metric_prefix(100, 1000)
(100.0, '')
>>> metric_prefix(999, 1000)
(999.0, '')
>>> metric_prefix(1000, 1000)
(1.0, 'K')
>>> metric_prefix(1001, 1000)
(1.001, 'K')
>>> metric_prefix(1000000, 1000)
(1.0, 'M')
>>> metric_prefix(1000**10, 1000)
(1.0, 'Q')
>>> metric_prefix(1000**11, 1000)
(1000.0, 'Q')

galaxy.util.shorten_with_metric_prefix(amount: int) → str

>>> shorten_with_metric_prefix(23000)
'23K'
>>> shorten_with_metric_prefix(2300000)
'2.3M'
>>> shorten_with_metric_prefix(23000000)
'23M'
>>> shorten_with_metric_prefix(1)
'1'
>>> shorten_with_metric_prefix(0)
'0'
>>> shorten_with_metric_prefix(100)
'100'
>>> shorten_with_metric_prefix(-100)
'-100'

galaxy.util.nice_size(size: float | int | str | Decimal) → str

Returns a readably formatted string with the size

>>> nice_size(100)
'100 bytes'
>>> nice_size(10000)
'9.8 KB'
>>> nice_size(1000000)
'976.6 KB'
>>> nice_size(100000000)
'95.4 MB'

galaxy.util.size_to_bytes(size)

Returns a number of bytes (as integer) if given a reasonably formatted string with the size

>>> size_to_bytes('1024')
1024
>>> size_to_bytes('1.0')
1
>>> size_to_bytes('10 bytes')
10
>>> size_to_bytes('4k')
4096
>>> size_to_bytes('2.2 TB')
2418925581107
>>> size_to_bytes('.01 TB')
10995116277
>>> size_to_bytes('1.b')
1
>>> size_to_bytes('1.2E2k')
122880

galaxy.util.send_mail(frm, to, subject, body, config, html=None, reply_to=None)

Sends an email.

Parameters:

• frm (str) – from address

• to (str) – to address

• subject (str) – Subject line

• body (str) – Body text (should be plain text)

• config (object) – Galaxy configuration object

• html (str) – Alternative HTML representation of the body content. If provided will convert the message to a MIMEMultipart. (Default None)

• reply_to (str) – Reply-to address (Default None)

galaxy.util.force_symlink(source, link_name)

galaxy.util.unlink(path_or_fd, ignore_errors=False)

Calls os.unlink on path_or_fd, and ignore FileNoteFoundError if ignore_errors is True.

galaxy.util.move_merge(source, target)

galaxy.util.safe_str_cmp(a, b)

safely compare two strings in a timing-attack-resistant manner

galaxy.util.galaxy_directory() → str

galaxy.util.in_packages() → bool

galaxy.util.config_directories_from_setting(directories_setting, galaxy_root=PosixPath('/home/runner/work/galaxy/galaxy'))

Parse the directories_setting into a list of relative or absolute filesystem paths that will be searched to discover plugins.

Parameters:

• galaxy_root (string) – the root path of this galaxy installation

• directories_setting (string (default: None)) – the filesystem path (or paths) to search for plugins. Can be CSV string of paths. Will be treated as absolute if a path starts with ‘/’, relative otherwise.

Return type:

list of strings

Returns:

list of filesystem paths

galaxy.util.parse_int(value, min_val=None, max_val=None, default=None, allow_none=False)

galaxy.util.parse_non_hex_float(s)

Parse string s into a float but throw a ValueError if the string is in the otherwise acceptable format d+ed+ (e.g. 40000000000000e5.)

This can be passed into json.loads to prevent a hex string in the above format from being incorrectly parsed as a float in scientific notation.

>>> parse_non_hex_float( '123.4' )
123.4
>>> parse_non_hex_float( '2.45e+3' )
2450.0
>>> parse_non_hex_float( '2.45e-3' )
0.00245
>>> parse_non_hex_float( '40000000000000e5' )
Traceback (most recent call last):
    ...
ValueError: could not convert string to float: 40000000000000e5

galaxy.util.build_url(base_url, port=80, scheme='http', pathspec=None, params=None, doseq=False)

galaxy.util.url_get(base_url, auth=None, pathspec=None, params=None, max_retries=5, backoff_factor=1)

Make contact with the uri provided and return any contents.

galaxy.util.is_url(uri, allow_list=None)

Check if uri is (most likely) an URL, more precisely the function checks if uri starts with a scheme from the allow list (defaults to “http://”, “https://”, “ftp://”) >>> is_url(’https://zenodo.org/record/4104428/files/UCSC-hg38-chr22-Coding-Exons.bed’) True >>> is_url(’file:///some/path’) False >>> is_url(‘/some/path’) False

galaxy.util.check_github_api_response_rate_limit(response)

galaxy.util.download_to_file(url, dest_file_path, timeout=30, chunk_size=1048576)

Download a URL to a file in chunks.

galaxy.util.stream_to_open_named_file(stream, fd, filename, source_encoding=None, source_error='strict', target_encoding=None, target_error='strict')

Writes a stream to the provided file descriptor, returns the file name. Closes file descriptor

class galaxy.util.classproperty(f)

Bases: object

__init__(f)

class galaxy.util.ExecutionTimer

Bases: object

__init__()

property elapsed

class galaxy.util.StructuredExecutionTimer(timer_id, template, **tags)

Bases: object

__init__(timer_id, template, **tags)

to_str(**kwd)

property elapsed

galaxy.util.enum_values(enum_class)

Return a list of member values of enumeration enum_class. Values are in member definition order.

galaxy.util.hex_to_lowercase_alphanum(hex_string: str) → str

Convert a hexadecimal string encoding into a lowercase 36-base alphanumeric string using the characters a-z and 0-9

galaxy.util.lowercase_alphanum_to_hex(lowercase_alphanum: str) → str

Convert a lowercase 36-base alphanumeric string encoding using the characters a-z and 0-9 to a hexadecimal string

Subpackages

• galaxy.util.custom_logging package
  • trace()
  • get_logger()
  • Submodules
  • galaxy.util.custom_logging.fluent_log module
    • FluentTraceLogger
      • FluentTraceLogger.__init__()
      • FluentTraceLogger.context_set()
      • FluentTraceLogger.context_remove()
      • FluentTraceLogger.log()
• galaxy.util.path package
  • get_ext()
  • has_ext()
  • join()
  • joinext()
  • safe_contains()
  • safe_makedirs()
  • safe_relpath()
  • safe_walk()
  • unsafe_walk()
  • Submodules
  • galaxy.util.path.ntpath module
  • galaxy.util.path.posixpath module
• galaxy.util.tool_shed package
  • Submodules
  • galaxy.util.tool_shed.common_util module
    • accumulate_tool_dependencies()
    • check_tool_tag_set()
    • generate_clone_url_for_installed_repository()
    • generate_clone_url_from_repo_info_tup()
    • get_repository_dependencies()
    • get_protocol_from_tool_shed_url()
    • get_tool_shed_repository_ids()
    • get_tool_shed_url_from_tool_shed_registry()
    • get_tool_shed_repository_url()
    • handle_galaxy_url()
    • handle_tool_shed_url_protocol()
    • parse_repository_dependency_tuple()
    • remove_port_from_tool_shed_url()
    • remove_protocol_and_port_from_tool_shed_url()
    • remove_protocol_and_user_from_clone_url()
    • remove_protocol_from_tool_shed_url()
  • galaxy.util.tool_shed.encoding_util module
    • tool_shed_decode()
    • tool_shed_encode()
  • galaxy.util.tool_shed.xml_util module
    • create_and_write_tmp_file()
    • parse_xml()

Submodules

galaxy.util.aliaspickler module

class galaxy.util.aliaspickler.AliasUnpickler(aliases, *args, **kw)

Bases: Unpickler

__init__(aliases, *args, **kw)

find_class(module, name)

Return an object from a specified module.

If necessary, the module will be imported. Subclasses may override this method (e.g. to restrict unpickling of arbitrary classes and functions).

This method is called whenever a class or a function object is needed. Both arguments passed are str objects.

class galaxy.util.aliaspickler.AliasPickleModule(aliases)

Bases: object

__init__(aliases)

dump(obj, fileobj, protocol=0)

dumps(obj, protocol=0)

load(fileobj)

loads(string)

galaxy.util.bool_expressions module

Simple boolean expression parser and evaluator using pyparsing.

Based on the example: https://github.com/pyparsing/pyparsing/blob/master/examples/simpleBool.py

class galaxy.util.bool_expressions.TokenEvaluator

Bases: object

Interface to evaluate a token and determine its boolean value.

evaluate(token: str) → bool

Returns the boolean representation of the given token according to some custom logic.

class galaxy.util.bool_expressions.BoolOperand(token)

Bases: object

Represents a boolean operand that has a label and a value.

The value is determined by a custom TokenEvaluator.

evaluator: TokenEvaluator

__init__(token)

class galaxy.util.bool_expressions.BoolBinaryOperation(token)

Bases: object

Base representation of a boolean binary operation.

reprsymbol: str

evalop: Callable[[Iterable[object]], bool]

__init__(token)

class galaxy.util.bool_expressions.BoolAnd(token)

Bases: BoolBinaryOperation

Represents the AND boolean operation.

reprsymbol: str = '&'

evalop()

Return True if bool(x) is True for all values x in the iterable.

If the iterable is empty, return True.

class galaxy.util.bool_expressions.BoolOr(token)

Bases: BoolBinaryOperation

Represents the OR boolean operation.

reprsymbol: str = '|'

evalop()

Return True if bool(x) is True for any x in the iterable.

If the iterable is empty, return False.

class galaxy.util.bool_expressions.BoolNot(token)

Bases: object

Represents the NOT boolean operation.

__init__(token)

class galaxy.util.bool_expressions.BooleanExpressionEvaluator(evaluator: TokenEvaluator, token_format: str | None = None)

Bases: object

Boolean logic parser that can evaluate an expression using a particular TokenEvaluator.

Supports AND, OR and NOT operator including parentheses to override operator precedences.

You can pass in different TokenEvaluator implementations to customize how the tokens (or variables) are converted to a boolean value when evaluating the expression.

__init__(evaluator: TokenEvaluator, token_format: str | None = None) → None

Initializes the expression evaluator.

Parameters:

• evaluator (TokenEvaluator) – The custom TokenEvaluator used to transform any token into a boolean.

• token_format (Optional[str]) – A string of all allowed characters used to form a valid token, defaults to None. The default value (None) will use DEFAULT_TOKEN_FORMAT which means the allowed characters are [A-Za-z0-9_-@.].

evaluate_expression(expr: str) → bool

Given an expression it gets evaluated to True or False using boolean logic.

classmethod is_valid_expression(expr: str) → bool

Tries to evaluate the given boolean expression and returns True if it is valid or False if it has syntax or gramatical errors.

class galaxy.util.bool_expressions.TokenContainedEvaluator(tokens: Set[str])

Bases: TokenEvaluator

Implements the TokenEvaluator interface to determine if a token is contained in a particular list of tokens.

__init__(tokens: Set[str]) → None

Initializes the token evaluator with the set of tokens that will evaluate to True.

Parameters:

tokens (List[str]) – The list of tokens that should be evaluated to True.

evaluate(token: str) → bool

Returns the boolean representation of the given token according to some custom logic.

class galaxy.util.bool_expressions.ValidationOnlyTokenEvaluator

Bases: TokenEvaluator

Simple TokenEvaluator that always evaluates to True for valid tokens.

This is only useful for validation purposes, do NOT use it for real expression evaluations.

evaluate(token: str) → bool

Returns the boolean representation of the given token according to some custom logic.

galaxy.util.bunch module

class galaxy.util.bunch.Bunch(**kwds)

Bases: object

http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/52308

Often we want to just collect a bunch of stuff together, naming each item of the bunch; a dictionary’s OK for that, but a small do-nothing class is even handier, and prettier to use.

__init__(**kwds)

dict()

get(key, default=None)

items()

keys()

values()

galaxy.util.bytesize module

class galaxy.util.bytesize.ByteSize(value)

Bases: object

Convert multiples of bytes to various units.

__init__(value)

Represents a quantity of bytes.

value may be an integer, in which case it is assumed to be bytes. If value is a string, it is parsed as bytes if no suffix (Mi, M, Gi, G …) is found.

>>> values = [128974848, '129e6', '129M', '123Mi' ]
>>> [ByteSize(v).to_unit('M') for v in values]
['128M', '129M', '129M', '128M']

to_unit(unit=None, as_string=True)

unit must be None or one of Ki,Mi,Gi,Ti,Pi,Ei,K,M,G,T,P.

galaxy.util.bytesize.parse_bytesize(value)

galaxy.util.checkers module

galaxy.util.checkers.check_binary(name, file_path: bool = True) → bool

galaxy.util.checkers.check_bz2(file_path: str, check_content: bool = True) → Tuple[bool, bool]

galaxy.util.checkers.check_gzip(file_path: str, check_content: bool = True) → Tuple[bool, bool]

galaxy.util.checkers.check_html(name, file_path: bool = True) → bool

Returns True if the file/string contains HTML code.

galaxy.util.checkers.check_image(file_path: str) → bool

Simple wrapper around image_type to yield a True/False verdict

galaxy.util.checkers.check_zip(file_path: str, check_content: bool = True, files=1) → Tuple[bool, bool]

galaxy.util.checkers.is_gzip(file_path: str) → bool

galaxy.util.checkers.is_bz2(file_path: str) → bool

galaxy.util.checkers.is_xz(file_path: str) → bool

galaxy.util.checkers.is_zip(file_path: str) → bool

galaxy.util.commands module

Generic I/O and shell processing code used by Galaxy tool dependencies.

galaxy.util.commands.argv_to_str(command_argv, quote=True)

Convert an argv command list to a string for shell subprocess.

If None appears in the command list it is simply excluded.

Arguments are quoted with shlex.quote(). That said, this method is not meant to be used in security critical paths of code and should not be used to sanitize code.

exception galaxy.util.commands.CommandLineException(command, stdout, stderr, returncode)

Bases: Exception

An exception indicating a non-zero command-line exit.

__init__(command, stdout, stderr, returncode)

Construct a CommandLineException from command and standard I/O.

galaxy.util.commands.download_command(url, to='-')

Build a command line to download a URL.

By default the URL will be downloaded to standard output but a specific file can be specified with the to argument.

galaxy.util.commands.execute(cmds, input=None)

Execute commands and throw an exception on a non-zero exit. if input is not None then the string is sent to the process’ stdin.

Return the standard output if the commands are successful

galaxy.util.commands.new_clean_env()

Returns a minimal environment to use when invoking a subprocess

galaxy.util.commands.redirect_aware_commmunicate(p, sys=<module 'sys' (built-in)>)

Variant of process.communicate that works with in process I/O redirection.

galaxy.util.commands.redirecting_io(sys=<module 'sys' (built-in)>)

Predicate to determine if we are redicting stdout in process.

galaxy.util.commands.shell(cmds: List[str] | str, env: Dict[str, str] | None = None, **kwds: Any) → int

Run shell commands with shell_process and wait.

galaxy.util.commands.shell_process(cmds: List[str] | str, env: Dict[str, str] | None = None, **kwds: Any) → Popen

A high-level method wrapping subprocess.Popen.

Handles details such as environment extension and in process I/O redirection.

galaxy.util.commands.which(file: str) → str | None

galaxy.util.compression_utils module

galaxy.util.compression_utils.get_fileobj(filename: str, mode: typing_extensions.Literal[r], compressed_formats: List[str] | None = None) → IO[str] | TextIOWrapper

galaxy.util.compression_utils.get_fileobj(filename: str, mode: typing_extensions.Literal[rb], compressed_formats: List[str] | None = None) → GzipFile | BZ2File | LZMAFile | IO[bytes]

galaxy.util.compression_utils.get_fileobj(filename: str) → IO[str] | TextIOWrapper

galaxy.util.compression_utils.get_fileobj(filename: str, mode: str = 'r', compressed_formats: List[str] | None = None) → IO[str] | TextIOWrapper | GzipFile | BZ2File | LZMAFile | IO[bytes]

Returns a fileobj. If the file is compressed, return an appropriate file reader. In text mode, always use ‘utf-8’ encoding.

Parameters:

• filename – path to file that should be opened

• mode – mode to pass to opener

• compressed_formats – list of allowed compressed file formats among ‘bz2’, ‘gzip’, ‘xz’ and ‘zip’. If left to None, all 3 formats are allowed

galaxy.util.compression_utils.get_fileobj_raw(filename: str, mode: typing_extensions.Literal[r], compressed_formats: List[str] | None = None) → Tuple[str | None, IO[str] | TextIOWrapper]

galaxy.util.compression_utils.get_fileobj_raw(filename: str, mode: typing_extensions.Literal[rb], compressed_formats: List[str] | None = None) → Tuple[str | None, GzipFile | BZ2File | LZMAFile | IO[bytes]]

galaxy.util.compression_utils.get_fileobj_raw(filename: str) → Tuple[str | None, IO[str] | TextIOWrapper]

galaxy.util.compression_utils.get_fileobj_raw(filename: str, mode: str = 'r', compressed_formats: List[str] | None = None) → Tuple[str | None, IO[str] | TextIOWrapper | GzipFile | BZ2File | LZMAFile | IO[bytes]]

galaxy.util.compression_utils.file_iter(fname: str, sep: Any | None = None) → Generator[List[bytes] | Any | List[str], None, None]

This generator iterates over a file and yields its lines splitted via the C{sep} parameter. Skips empty lines and lines starting with the C{#} character.

>>> lines = [ line for line in file_iter(__file__) ]
>>> len(lines) !=  0
True

galaxy.util.compression_utils.decompress_bytes_to_directory(content: bytes) → str

galaxy.util.compression_utils.decompress_path_to_directory(path: str) → str

class galaxy.util.compression_utils.CompressedFile(file_path: str | PathLike, mode: str = 'r')

Bases: object

static can_decompress(file_path: str | PathLike) → bool

__init__(file_path: str | PathLike, mode: str = 'r') → None

archive: TarFile | ZipFile

property common_prefix_dir: str

Get the common prefix directory for all the files in the archive, if any.

Returns ‘’ if the archive contains multiple files and/or directories at the root of the archive.

extract(path: str | PathLike) → str

Determine the path to which the archive should be extracted.

safemembers() → Iterable[TarInfo] | Iterable[str]

getmembers_tar() → List[TarInfo]

getmembers_zip() → List[ZipInfo]

getname_tar(item: TarInfo) → str

getname_zip(item: ZipInfo) → str

getmember(name: str) → TarInfo | ZipInfo | None

getmembers() → List[TarInfo | ZipInfo]

getname(member: TarInfo | ZipInfo) → str

isdir(member: TarInfo | ZipInfo) → bool

isdir_tar(member: TarInfo) → bool

isdir_zip(member: ZipInfo) → bool

isfile(member: TarInfo | ZipInfo) → bool

open_tar(filepath: str | PathLike, mode: str) → TarFile

open_zip(filepath: str | PathLike, mode: str) → ZipFile

zipfile_ok(path_to_archive: str | PathLike) → bool

This function is a bit pedantic and not functionally necessary. It checks whether there is no file pointing outside of the extraction, because ZipFile.extractall() has some potential security holes. See python zipfile documentation for more details.

class galaxy.util.compression_utils.FastZipFile(file, mode='r', compression=0, allowZip64=True, compresslevel=None, *, strict_timestamps=True)

Bases: ZipFile

Simple wrapper around ZipFile that uses the default compression strategy of ISA-L to write zip files. Ignores compresslevel and compresstype arguments, and is 3 to 4 times faster than the zlib implementation at the default compression level.

galaxy.util.compression_utils.make_fast_zipfile(base_name: str, base_dir: str, verbose: int = 0, dry_run: int = 0, logger: Logger | None = None, owner: str | None = None, group: str | None = None, root_dir: str | None = None) → str

Create a zip file from all the files under ‘base_dir’.

The output zip file will be named ‘base_name’ + “.zip”. Returns the name of the output zip file.

galaxy.util.dbkeys module

galaxy.util.dictifiable module

galaxy.util.dictifiable.dict_for(obj, **kwds)

class galaxy.util.dictifiable.Dictifiable

Bases: object

Mixin that enables objects to be converted to dictionaries. This is useful when for sharing objects across boundaries, such as the API, tool scripts, and JavaScript code.

to_dict(view: str = 'collection', value_mapper: Dict[str, Callable] | None = None) → Dict[str, Any]

Return item dictionary.

galaxy.util.dynamic module

galaxy.util.expressions module

Expression evaluation support.

For the moment this depends on python’s eval. In the future it should be replaced with a “safe” parser.

class galaxy.util.expressions.ExpressionContext(dict, parent=None)

Bases: MutableMapping

__init__(dict, parent=None)

Create a new expression context that looks for values in the container object ‘dict’, and falls back to ‘parent’

galaxy.util.facts module

Return various facts for string formatting.

class galaxy.util.facts.Facts(config=None, **kwargs)

Bases: MutableMapping

A dict-like object that evaluates values at access time.

__init__(config=None, **kwargs)

galaxy.util.facts.get_facts(config=None, **kwargs)

galaxy.util.filelock module

Code obtained from https://github.com/dmfrey/FileLock.

See full license at:

https://github.com/dmfrey/FileLock/blob/master/LICENSE.txt

exception galaxy.util.filelock.FileLockException

Bases: Exception

class galaxy.util.filelock.FileLock(file_name, timeout=10, delay=0.05)

Bases: object

A file locking mechanism that has context-manager support so you can use it in a with statement. This should be relatively cross compatible as it doesn’t rely on msvcrt or fcntl for the locking.

__init__(file_name, timeout=10, delay=0.05)

Prepare the file locker. Specify the file to lock and optionally the maximum timeout and the delay between each attempt to lock.

acquire()

Acquire the lock, if possible. If the lock is in use, it check again every wait seconds. It does this until it either gets the lock or exceeds timeout number of seconds, in which case it throws an exception.

release()

Get rid of the lock by deleting the lockfile. When working in a with statement, this gets automatically called at the end.

galaxy.util.form_builder module

Classes for generating HTML forms

class galaxy.util.form_builder.BaseField(name, value=None, label=None, **kwds)

Bases: object

__init__(name, value=None, label=None, **kwds)

to_dict()

class galaxy.util.form_builder.TextField(name, value=None, label=None, **kwds)

Bases: BaseField

A standard text input box.

to_dict()

class galaxy.util.form_builder.PasswordField(name, value=None, label=None, **kwds)

Bases: BaseField

A password input box. text appears as “**”

to_dict()

class galaxy.util.form_builder.TextArea(name, value=None, label=None, **kwds)

Bases: BaseField

A standard text area box.

to_dict()

class galaxy.util.form_builder.CheckboxField(name, value=None, label=None, **kwds)

Bases: BaseField

A checkbox (boolean input)

static is_checked(value)

to_dict()

class galaxy.util.form_builder.SelectField(name, multiple=None, display=None, field_id=None, value=None, selectlist=None, refresh_on_change=False, **kwds)

Bases: BaseField

A select field.

__init__(name, multiple=None, display=None, field_id=None, value=None, selectlist=None, refresh_on_change=False, **kwds)

add_option(label, value, selected=False)

to_dict()

class galaxy.util.form_builder.AddressField(name, user=None, value=None, security=None, **kwds)

Bases: BaseField

static fields()

__init__(name, user=None, value=None, security=None, **kwds)

to_dict()

class galaxy.util.form_builder.WorkflowField(name, user=None, value=None, security=None, **kwds)

Bases: BaseField

__init__(name, user=None, value=None, security=None, **kwds)

to_dict()

class galaxy.util.form_builder.WorkflowMappingField(name, user=None, value=None, **kwds)

Bases: BaseField

__init__(name, user=None, value=None, **kwds)

class galaxy.util.form_builder.HistoryField(name, user=None, value=None, security=None, **kwds)

Bases: BaseField

__init__(name, user=None, value=None, security=None, **kwds)

to_dict()

galaxy.util.hash_util module

Utility functions for bi-directional Python version compatibility. Python 2.5 introduced hashlib which replaced sha in Python 2.4 and previous versions.

galaxy.util.hash_util.md5(string=b'')

Returns a md5 hash object; optionally initialized with a string

galaxy.util.hash_util.sha1(string=b'')

Returns a sha1 hash object; optionally initialized with a string

galaxy.util.hash_util.sha(string=b'')

Returns a sha1 hash object; optionally initialized with a string

galaxy.util.hash_util.new_insecure_hash(text_type: bytes | str) → str

Returns the hexdigest of the sha1 hash of the argument text_type.

Previously called new_secure_hash, but this should not be considered secure - SHA1 is no longer considered a secure hash and has been broken since the early 2000s.

use_pbkdf2 should be set by default and galaxy.security.passwords should be the default used for passwords in Galaxy.

galaxy.util.hash_util.new_secure_hash_v2(text_type: bytes | str) → str

More modern version of new_secure_hash.

Certain passwords are set via new_insecure_hash (previously new_secure_hash), so that needs to remain for legacy purposes.

galaxy.util.hash_util.hmac_new(key: bytes | str, value: bytes | str) → str

galaxy.util.hash_util.is_hashable(value: Any) → bool

galaxy.util.hash_util.parse_checksum_hash(checksum: str) → Tuple[HashFunctionNameEnum, str]

Parses checksum strings in the form of hash_type$hash_value considering possible aliases.

galaxy.util.heartbeat module

galaxy.util.heartbeat.get_current_thread_object_dict()

Get a dictionary of all ‘Thread’ objects created via the threading module keyed by thread_id. Note that not all interpreter threads have a thread objects, only the main thread and any created via the ‘threading’ module. Threads created via the low level ‘thread’ module will not be in the returned dictionary.

HACK: This mucks with the internals of the threading module since that

module does not expose any way to match ‘Thread’ objects with intepreter thread identifiers (though it should).

class galaxy.util.heartbeat.Heartbeat(config, name='Heartbeat Thread', period=20, fname='heartbeat.log')

Bases: Thread

Thread that periodically dumps the state of all threads to a file

__init__(config, name='Heartbeat Thread', period=20, fname='heartbeat.log')

This constructor should always be called with keyword arguments. Arguments are:

group should be None; reserved for future extension when a ThreadGroup class is implemented.

target is the callable object to be invoked by the run() method. Defaults to None, meaning nothing is called.

name is the thread name. By default, a unique name is constructed of the form “Thread-N” where N is a small decimal number.

args is the argument tuple for the target invocation. Defaults to ().

kwargs is a dictionary of keyword arguments for the target invocation. Defaults to {}.

If a subclass overrides the constructor, it must make sure to invoke the base class constructor (Thread.__init__()) before doing anything else to the thread.

run()

Method representing the thread’s activity.

You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.

open_logs()

close_logs()

dump()

shutdown()

thread_is_sleeping(last_stack_frame)

Returns True if the given stack-frame represents a known sleeper function (at least in python 2.5)

get_interesting_stack_frame(stack_frames)

Scans a given backtrace stack frames, returns a single quadraple of [filename, line, function-name, text] of the single, deepest, most interesting frame.

Interesting being:

inside the galaxy source code ("/lib/galaxy"),
prefreably not an egg.

print_nonsleeping(threads_object_dict)

dump_signal_handler(signum, frame)

galaxy.util.image_util module

Provides utilities for working with image files.

galaxy.util.image_util.image_type(filename: str) → str | None

galaxy.util.image_util.check_image_type(filename: str, types: List[str]) → bool

galaxy.util.inflection module

class galaxy.util.inflection.Inflector

Bases: object

Inflector for pluralizing and singularizing English nouns.

NONCHANGING_WORDS = {'equipment', 'information', 'money', 'rice', 'series', 'sheep', 'sms', 'species'}

IRREGULAR_WORDS = {'child': 'children', 'man': 'men', 'move': 'moves', 'octopus': 'octopi', 'person': 'people', 'sex': 'sexes'}

PLURALIZE_RULES = (('(?i)(quiz)$', '\\1zes'), ('(?i)^(ox)$', '\\1en'), ('(?i)([m|l])ouse$', '\\1ice'), ('(?i)(matr|vert|ind)ix|ex$', '\\1ices'), ('(?i)(x|ch|ss|sh)$', '\\1es'), ('(?i)([^aeiouy]|qu)ies$', '\\1y'), ('(?i)([^aeiouy]|qu)y$', '\\1ies'), ('(?i)(hive)$', '\\1s'), ('(?i)(?:([^f])fe|([lr])f)$', '\\1\\2ves'), ('(?i)sis$', 'ses'), ('(?i)([ti])um$', '\\1a'), ('(?i)(buffal|tomat)o$', '\\1oes'), ('(?i)(bu)s$', '\\1ses'), ('(?i)(alias|status|virus)', '\\1es'), ('(?i)(ax|test)is$', '\\1es'), ('(?i)s$', 's'), ('(?i)$', 's'))

SINGULARIZE_RULES = (('(?i)(quiz)zes$', '\\1'), ('(?i)(matr)ices$', '\\1ix'), ('(?i)(vert|ind)ices$', '\\1ex'), ('(?i)^(ox)en', '\\1'), ('(?i)(alias|status|virus)es$', '\\1'), ('(?i)(cris|ax|test)es$', '\\1is'), ('(?i)(shoe)s$', '\\1'), ('(?i)(o)es$', '\\1'), ('(?i)(bus)es$', '\\1'), ('(?i)([m|l])ice$', '\\1ouse'), ('(?i)(x|ch|ss|sh)es$', '\\1'), ('(?i)(m)ovies$', '\\1ovie'), ('(?i)(s)eries$', '\\1eries'), ('(?i)([^aeiouy]|qu)ies$', '\\1y'), ('(?i)([lr])ves$', '\\1f'), ('(?i)(tive)s$', '\\1'), ('(?i)(hive)s$', '\\1'), ('(?i)([^f])ves$', '\\1fe'), ('(?i)(^analy)ses$', '\\1sis'), ('(?i)((a)naly|(b)a|(d)iagno|(p)arenthe|(p)rogno|(s)ynop|(t)he)ses$', '\\1\\2sis'), ('(?i)([ti])a$', '\\1um'), ('(?i)(n)ews$', '\\1ews'), ('(?i)s$', ''))

pluralize(word)

Pluralizes nouns.

singularize(word)

Singularizes nouns.

cond_plural(number_of_records, word)

Returns the plural form of a word if first parameter is greater than 1

galaxy.util.json module

galaxy.util.json.safe_dumps(obj, **kwargs)

This is a wrapper around dumps that encodes Infinity and NaN values. It’s a fairly rare case (which will be low in request volume). Basically, we tell json.dumps to blow up if it encounters Infinity/NaN, or Decimal values and we ‘fix’ it before re-encoding.

galaxy.util.json.validate_jsonrpc_request(request, regular_methods, notification_methods)

galaxy.util.json.validate_jsonrpc_response(response, id=None)

galaxy.util.json.jsonrpc_request(method, params=None, id=None, jsonrpc='2.0')

galaxy.util.json.jsonrpc_response(request=None, id=None, result=None, error=None, jsonrpc='2.0')

galaxy.util.jstree module

class galaxy.util.jstree.Path(path, id, options)

Bases: tuple

id

Alias for field number 1

options

Alias for field number 2

path

Alias for field number 0

class galaxy.util.jstree.Node(path, oid, **kwargs)

Bases: DictionaryObject

Copyright 2012 “Grim Apps”

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

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Helper class written by William Grim - grimwm Original repo: https://github.com/grimwm/py-jstree Code adjusted according to the idea of Frank Blechschmidt - FraBle Thank you! ******************** This class exists as a helper to the JSTree. Its “jsonData” method can generate sub-tree JSON without putting the logic directly into the JSTree.

This data structure is only semi-immutable. The JSTree uses a directly iterative (i.e. no stack is managed) builder pattern to construct a tree out of paths. Therefore, the children are not known in advance, and we have to keep the children attribute mutable.

__init__(path, oid, **kwargs)

kwargs allows users to pass arbitrary information into a Node that will later be output in jsonData(). It allows for more advanced configuration than the default path handling that JSTree currently allows. For example, users may want to pass “attr” or some other valid jsTree options.

Example:

>>> node = Node('a', None)
>>> assert node._items == {'text': 'a', 'children': dictobj.MutableDictionaryObject({})}
>>> assert node.jsonData() == {'text': 'a'}

>>> node = Node('a', 1)
>>> assert node._items == {'text': 'a', 'children': dictobj.MutableDictionaryObject({}), 'li_attr': dictobj.DictionaryObject({'id': 1}), 'id': 1}
>>> assert node.jsonData() == {'text': 'a', 'id': 1, 'li_attr': {'id': 1}}

>>> node = Node('a', 5, icon="folder", state = {'opened': True})
>>> assert node._items == {'text': 'a', 'id': 5, 'state': dictobj.DictionaryObject({'opened': True}), 'children': dictobj.MutableDictionaryObject({}), 'li_attr': dictobj.DictionaryObject({'id': 5}), 'icon': 'folder'}
>>> assert node.jsonData() == {'text': 'a', 'state': {'opened': True}, 'id': 5, 'li_attr': {'id': 5}, 'icon': 'folder'}

jsonData()

class galaxy.util.jstree.JSTree(paths, **kwargs)

Bases: DictionaryObject

An immutable dictionary-like object that converts a list of “paths” into a tree structure suitable for jQuery’s jsTree.

__init__(paths, **kwargs)

Take a list of paths and put them into a tree. Paths with the same prefix should be at the same level in the tree.

kwargs may be standard jsTree options used at all levels in the tree. These will be outputted in the JSON.

pretty(root=None, depth=0, spacing=2)

Create a “pretty print” represenation of the tree with customized indentation at each level of the tree.

jsonData()

Returns a copy of the internal tree in a JSON-friendly format, ready for consumption by jsTree. The data is represented as a list of dictionaries, each of which are our internal nodes.

galaxy.util.lazy_process module

class galaxy.util.lazy_process.LazyProcess(command_and_args)

Bases: object

Abstraction describing a command line launching a service - probably as needed as functionality is accessed in Galaxy.

__init__(command_and_args)

start_process()

shutdown()

property running

class galaxy.util.lazy_process.NoOpLazyProcess

Bases: object

LazyProcess abstraction meant to describe potentially optional services, in those cases where one is not configured or valid, this class can be used in place of LazyProcess.

start_process()

shutdown()

property running

galaxy.util.monitors module

class galaxy.util.monitors.Monitors

Bases: object

start_monitoring()

stop_monitoring()

shutdown_monitor()

galaxy.util.odict module

Ordered dictionary implementation with insert functionality.

This is only used in one specific place in the codebase:

galaxy.tool_util.toolbox.panel

Whenever possible the stdlib collections.OrderedDict should be used instead of this custom implementation.

class galaxy.util.odict.odict(dict=None)

Bases: UserDict

http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/107747

This dictionary class extends UserDict to record the order in which items are added. Calling keys(), values(), items(), etc. will return results in this order.

__init__(dict=None)

clear() → None.  Remove all items from D.

copy()

items() → a set-like object providing a view on D's items

keys() → a set-like object providing a view on D's keys

popitem() → (k, v), remove and return some (key, value) pair

as a 2-tuple; but raise KeyError if D is empty.

setdefault(k[, d]) → D.get(k,d), also set D[k]=d if k not in D

update([E, ]**F) → None.  Update D from mapping/iterable E and F.

If E present and has a .keys() method, does: for k in E: D[k] = E[k] If E present and lacks .keys() method, does: for (k, v) in E: D[k] = v In either case, this is followed by: for k, v in F.items(): D[k] = v

values() → an object providing a view on D's values

iterkeys()

itervalues()

iteritems()

reverse()

insert(index, key, item)

galaxy.util.oset module

Ordered set implementation from https://code.activestate.com/recipes/576694/

class galaxy.util.oset.OrderedSet(iterable=None)

Bases: MutableSet

__init__(iterable=None)

add(key)

Add an element.

discard(key)

Remove an element. Do not raise an exception if absent.

pop(last=True)

Return the popped value. Raise KeyError if empty.

galaxy.util.permutations module

There is some shared logic between matching/multiplying inputs in workflows and tools. This module is meant to capture some general permutation logic that can be applicable for both cases but will only be used in the newer tools case first.

Maybe this doesn’t make sense and maybe much of this stuff could be replaced with itertools product and permutations. These are open questions.

exception galaxy.util.permutations.InputMatchedException(err_msg: str | None = None, type='info', **extra_error_info)

Bases: MessageException

Indicates problem matching inputs while building up inputs permutations.

galaxy.util.permutations.expand_multi_inputs(inputs: Dict[str, T], classifier, key_filter=None)

galaxy.util.plugin_config module

class galaxy.util.plugin_config.PluginConfigSource(type, source)

Bases: tuple

type: str

Alias for field number 0

source: Any

Alias for field number 1

galaxy.util.plugin_config.plugins_dict(module: module, plugin_type_identifier: str) → Dict[str, Type]

Walk through all classes in submodules of module and find ones labelled with specified plugin_type_identifier and throw in a dictionary to allow constructions from plugins by these types later on.

galaxy.util.plugin_config.load_plugins(plugins_dict: Dict[str, Type], plugin_source: PluginConfigSource, extra_kwds: Dict[str, Any] | None = None, plugin_type_keys: Iterable[str] = ('type',), dict_to_list_key: str | None = None) → List[Any]

galaxy.util.plugin_config.plugin_source_from_path(path: str | Path) → PluginConfigSource

galaxy.util.plugin_config.plugin_source_from_dict(as_dict: Dict[str, Any] | List[Dict[str, Any]]) → PluginConfigSource

galaxy.util.properties module

Module used to blend ini, environment, and explicit dictionary properties to determine application configuration. Some hard coded defaults for Galaxy but this should be reusable by tool shed and pulsar as well.

galaxy.util.properties.find_config_file(names, exts=None, dirs=None, include_samples=False)

Locate a config file in multiple directories, with multiple extensions.

>>> from shutil import rmtree
>>> from tempfile import mkdtemp
>>> def touch(d, f):
...     open(os.path.join(d, f), 'w').close()
>>> def _find_config_file(*args, **kwargs):
...     return find_config_file(*args, **kwargs).replace(d, '')
>>> d = mkdtemp()
>>> d1 = os.path.join(d, 'd1')
>>> d2 = os.path.join(d, 'd2')
>>> os.makedirs(d1)
>>> os.makedirs(d2)
>>> touch(d1, 'foo.ini')
>>> touch(d1, 'foo.bar')
>>> touch(d1, 'baz.ini.sample')
>>> touch(d2, 'foo.yaml')
>>> touch(d2, 'baz.yml')
>>> _find_config_file('foo', dirs=(d1, d2))
'/d1/foo.ini'
>>> _find_config_file('baz', dirs=(d1, d2))
'/d2/baz.yml'
>>> _find_config_file('baz', dirs=(d1, d2), include_samples=True)
'/d2/baz.yml'
>>> _find_config_file('baz', dirs=(d1,), include_samples=True)
'/d1/baz.ini.sample'
>>> _find_config_file('foo', dirs=(d2, d1))
'/d2/foo.yaml'
>>> find_config_file('quux', dirs=(d,))
>>> _find_config_file('foo', exts=('bar', 'ini'), dirs=(d1,))
'/d1/foo.bar'
>>> rmtree(d)

galaxy.util.properties.get_data_dir(properties)

galaxy.util.properties.load_app_properties(kwds=None, ini_file=None, ini_section=None, config_file=None, config_section=None, config_prefix='GALAXY_CONFIG_')

class galaxy.util.properties.NicerConfigParser(filename, *args, **kw)

Bases: ConfigParser

__init__(filename, *args, **kw)

optionxform(optionstr: str) → str

defaults()

Return the defaults, with their values interpolated (with the defaults dict itself)

Mainly to support defaults using values such as %(here)s

galaxy.util.renamed_temporary_file module

Safely write file to temporary file and then move file into place.

class galaxy.util.renamed_temporary_file.RenamedTemporaryFile(final_path: str | bytes | PathLike, **kwargs)

Bases: object

A temporary file object which will be renamed to the specified path on exit.

__init__(final_path: str | bytes | PathLike, **kwargs)

>>> dir = tempfile.mkdtemp()
>>> with RenamedTemporaryFile(os.path.join(dir, 'test.txt'), mode="w") as out:
...     _ = out.write('bla')

final_path: str | bytes | PathLike

galaxy.util.resources module

Provide a consistent interface into and utilities for importlib file resources.

galaxy.util.resources.as_file(path)

galaxy.util.resources.as_file(path: Path)

Given a Traversable object, return that object as a path on the local file system in a context manager.

galaxy.util.resources.files(anchor: module | str | None = None) → Traversable

Get a Traversable resource for an anchor.

galaxy.util.resources.resource_string(package_or_requirement: module | str, resource_name: str) → str

Return specified resource as a string.

Replacement function for pkg_resources.resource_string, but returns unicode string instead of bytestring.

galaxy.util.resources.resource_path(package_or_requirement: module | str, resource_name: str) → Traversable

Return specified resource as a Traversable.

class galaxy.util.resources.Traversable(*args, **kwargs)

Bases: Protocol

An object with a subset of pathlib.Path methods suitable for traversing directories and opening files.

Any exceptions that occur when accessing the backing resource may propagate unaltered.

abstract iterdir() → Iterator[Traversable]

Yield Traversable objects in self

read_bytes() → bytes

Read contents of self as bytes

read_text(encoding: str | None = None) → str

Read contents of self as text

abstract is_dir() → bool

Return True if self is a directory

abstract is_file() → bool

Return True if self is a file

joinpath(*descendants: str | os.PathLike[str]) → Traversable

Return Traversable resolved with any descendants applied.

Each descendant should be a path segment relative to self and each may contain multiple levels separated by posixpath.sep (/).

abstract open(mode='r', *args, **kwargs)

mode may be ‘r’ or ‘rb’ to open as text or binary. Return a handle suitable for reading (same as pathlib.Path.open).

When opening as text, accepts encoding parameters such as those accepted by io.TextIOWrapper.

abstract property name: str

The base name of this object without any parent references.

__init__(*args, **kwargs)

galaxy.util.rules_dsl module

galaxy.util.rules_dsl.get_rules_specification()

galaxy.util.rules_dsl.apply_regex(regex, target, data, replacement=None, group_count=None)

class galaxy.util.rules_dsl.BaseRuleDefinition

Bases: object

abstract property rule_type

Short string describing type of rule (plugin class) to use.

abstract validate_rule(rule)

Validate dictified rule definition of this type.

abstract apply(rule, data, sources)

Apply validated, dictified rule definition to supplied data.

class galaxy.util.rules_dsl.AddColumnMetadataRuleDefinition

Bases: BaseRuleDefinition

rule_type = 'add_column_metadata'

validate_rule(rule)

Validate dictified rule definition of this type.

apply(rule, data, sources)

Apply validated, dictified rule definition to supplied data.

class galaxy.util.rules_dsl.AddColumnGroupTagValueRuleDefinition

Bases: BaseRuleDefinition

rule_type = 'add_column_group_tag_value'

validate_rule(rule)

Validate dictified rule definition of this type.

apply(rule, data, sources)

Apply validated, dictified rule definition to supplied data.

class galaxy.util.rules_dsl.AddColumnConcatenateRuleDefinition

Bases: BaseRuleDefinition

rule_type = 'add_column_concatenate'

validate_rule(rule)

Validate dictified rule definition of this type.

apply(rule, data, sources)

Apply validated, dictified rule definition to supplied data.

class galaxy.util.rules_dsl.AddColumnBasenameRuleDefinition

Bases: BaseRuleDefinition

rule_type = 'add_column_basename'

validate_rule(rule)

Validate dictified rule definition of this type.

apply(rule, data, sources)

Apply validated, dictified rule definition to supplied data.

class galaxy.util.rules_dsl.AddColumnRegexRuleDefinition

Bases: BaseRuleDefinition

rule_type = 'add_column_regex'

validate_rule(rule)

Validate dictified rule definition of this type.

apply(rule, data, sources)

Apply validated, dictified rule definition to supplied data.

class galaxy.util.rules_dsl.AddColumnRownumRuleDefinition

Bases: BaseRuleDefinition

rule_type = 'add_column_rownum'

validate_rule(rule)

Validate dictified rule definition of this type.

apply(rule, data, sources)

Apply validated, dictified rule definition to supplied data.

class galaxy.util.rules_dsl.AddColumnValueRuleDefinition

Bases: BaseRuleDefinition

rule_type = 'add_column_value'

validate_rule(rule)

Validate dictified rule definition of this type.

apply(rule, data, sources)

Apply validated, dictified rule definition to supplied data.

class galaxy.util.rules_dsl.AddColumnSubstrRuleDefinition

Bases: BaseRuleDefinition

rule_type = 'add_column_substr'

validate_rule(rule)

Validate dictified rule definition of this type.

apply(rule, data, sources)

Apply validated, dictified rule definition to supplied data.

class galaxy.util.rules_dsl.RemoveColumnsRuleDefinition

Bases: BaseRuleDefinition

rule_type = 'remove_columns'

validate_rule(rule)

Validate dictified rule definition of this type.

apply(rule, data, sources)

Apply validated, dictified rule definition to supplied data.

class galaxy.util.rules_dsl.AddFilterRegexRuleDefinition

Bases: BaseRuleDefinition

rule_type = 'add_filter_regex'

validate_rule(rule)

Validate dictified rule definition of this type.

apply(rule, data, sources)

Apply validated, dictified rule definition to supplied data.

class galaxy.util.rules_dsl.AddFilterCountRuleDefinition

Bases: BaseRuleDefinition

rule_type = 'add_filter_count'

validate_rule(rule)

Validate dictified rule definition of this type.

apply(rule, data, sources)

Apply validated, dictified rule definition to supplied data.

class galaxy.util.rules_dsl.AddFilterEmptyRuleDefinition

Bases: BaseRuleDefinition

rule_type = 'add_filter_empty'

validate_rule(rule)

Validate dictified rule definition of this type.

apply(rule, data, sources)

Apply validated, dictified rule definition to supplied data.

class galaxy.util.rules_dsl.AddFilterMatchesRuleDefinition

Bases: BaseRuleDefinition

rule_type = 'add_filter_matches'

validate_rule(rule)

Validate dictified rule definition of this type.

apply(rule, data, sources)

Apply validated, dictified rule definition to supplied data.

class galaxy.util.rules_dsl.AddFilterCompareRuleDefinition

Bases: BaseRuleDefinition

rule_type = 'add_filter_compare'

validate_rule(rule)

Validate dictified rule definition of this type.

apply(rule, data, sources)

Apply validated, dictified rule definition to supplied data.

class galaxy.util.rules_dsl.SortRuleDefinition

Bases: BaseRuleDefinition

rule_type = 'sort'

validate_rule(rule)

Validate dictified rule definition of this type.

apply(rule, data, sources)

Apply validated, dictified rule definition to supplied data.

class galaxy.util.rules_dsl.SwapColumnsRuleDefinition

Bases: BaseRuleDefinition

rule_type = 'swap_columns'

validate_rule(rule)

Validate dictified rule definition of this type.

apply(rule, data, sources)

Apply validated, dictified rule definition to supplied data.

class galaxy.util.rules_dsl.SplitColumnsRuleDefinition

Bases: BaseRuleDefinition

rule_type = 'split_columns'

validate_rule(rule)

Validate dictified rule definition of this type.

apply(rule, data, sources)

Apply validated, dictified rule definition to supplied data.

galaxy.util.rules_dsl.flat_map(f, items)

class galaxy.util.rules_dsl.RuleSet(rule_set_as_dict)

Bases: object

__init__(rule_set_as_dict)

property rules

apply(data, sources)

property has_errors

property mapping_as_dict

property identifier_columns

property collection_type

property display

galaxy.util.rules_dsl.rule_class

alias of SplitColumnsRuleDefinition

galaxy.util.sanitize_html module

HTML Sanitizer (lists of acceptable_* ripped from feedparser)

galaxy.util.sanitize_html.sanitize_html(htmlSource, allow_data_urls=False)

galaxy.util.script module

Utilities for Galaxy scripts

galaxy.util.script.main_factory(description=None, actions=None, arguments=None, default_action=None)

galaxy.util.script.main(argv=None)

Entry point for conversion process.

galaxy.util.script.app_properties_from_args(args, legacy_config_override=None, app=None)

galaxy.util.script.config_file_from_args(args, legacy_config_override=None, app=None)

galaxy.util.script.populate_config_args(parser)

galaxy.util.script.set_log_handler(filename=None, stream=None)

galaxy.util.search module

galaxy.util.search.parse_filters(search_term: str, filters: Dict[str, str] | None = None) → Tuple[List[FilteredTerm] | None, str | None]

Support github-like filters for narrowing the results.

Order of chunks does not matter, only recognized filter names are allowed.

Parameters:

search_term – the original search str from user input

Returns allow_query:

whoosh Query object used for filtering results of searching in index

Returns search_term_without_filters:

str that represents user’s search phrase without the filters

galaxy.util.search.parse_filters_structured(search_term: str, filters: Dict[str, str] | None = None, preserve_quotes: bool = True) → ParsedSearch

galaxy.util.simplegraph module

Fencepost-simple graph structure implementation.

class galaxy.util.simplegraph.SimpleGraphNode(index, **data)

Bases: object

Node representation.

__init__(index, **data)

Parameters:

• index (int) – index of this node in some parent list

• data ((variadic dictionary)) – any extra data that needs to be saved

class galaxy.util.simplegraph.SimpleGraphEdge(source_index, target_index, **data)

Bases: object

Edge representation.

__init__(source_index, target_index, **data)

Parameters:

• source_index (int) – index of the edge’s source node in some parent list

• target_index (int) – index of the edge’s target node in some parent list

• data ((variadic dictionary)) – any extra data that needs to be saved

class galaxy.util.simplegraph.SimpleGraph(nodes=None, edges=None)

Bases: object

Each node is unique (by id) and stores its own index in the node list/odict. Each edge is represented as two indeces into the node list/odict. Both nodes and edges allow storing extra information if needed.

Allows:

multiple edges between two nodes self referential edges (an edge from a node to itself)

These graphs are not specifically directed but since source and targets on the edges are listed - it could easily be used that way.

__init__(nodes=None, edges=None)

add_node(node_id, **data)

Adds a new node only if it doesn’t already exist. :param node_id: some unique identifier :type node_id: (hashable) :param data: any extra data that needs to be saved :type data: (variadic dictionary) :returns: the new node

add_edge(source_id, target_id, **data)

Adds a new node only if it doesn’t already exist. :param source_id: the id of the source node :type source_id: (hashable) :param target_id: the id of the target node :type target_id: (hashable) :param data: any extra data that needs to be saved for the edge :type data: (variadic dictionary) :returns: the new node

..note: that, although this will create new nodes if necessary, there’s no way to pass data to them - so if you need to assoc. more data with the nodes, use add_node first.

gen_node_dicts()

Returns a generator that yields node dictionaries in the form:

{ ‘id’: <the nodes unique id>, ‘data’: <any additional node data> }

gen_edge_dicts()

Returns a generator that yields node dictionaries in the form:

{
    'source': <the index of the source node in the graph's node list>,
    'target': <the index of the target node in the graph's node list>,
    'data'  : <any additional edge data>
}

as_dict()

Returns a dictionary of the form:

{ 'nodes': <a list of node dictionaries>, 'edges': <a list of node dictionaries> }

galaxy.util.sleeper module

class galaxy.util.sleeper.Sleeper

Bases: object

Provides a ‘sleep’ method that sleeps for a number of seconds unless the notify method is called (from a different thread).

__init__()

sleep(seconds)

wake()

galaxy.util.sockets module

galaxy.util.sockets.get_ip() → str

galaxy.util.sockets.unused_port(range=None)

galaxy.util.specs module

galaxy.util.specs.to_str_or_none(value)

galaxy.util.specs.to_bool_or_none(value)

galaxy.util.specs.to_bool(value)

galaxy.util.specs.to_float_or_none(value)

galaxy.util.specs.is_in(*args)

galaxy.util.sqlite module

galaxy.util.sqlite.is_read_only_query(query)

galaxy.util.sqlite.connect(path)

galaxy.util.submodules module

galaxy.util.submodules.import_submodules(module, ordered=True, recursive=False)

Import all submodules of a module

Parameters:

• module (str | module) – module (package name or actual module)

• ordered (bool) – Whether to order the returned modules. The default is True, and modules are returned in reverse order to allow hierarchical overrides i.e. 000_galaxy_rules.py, 100_site_rules.py, 200_instance_rules.py

• recursive (bool) – Recursively returns all subpackages

Return type:

[module]

galaxy.util.task module

class galaxy.util.task.IntervalTask(func, name='Periodic task', interval=3600, immediate_start=False, time_execution=False)

Bases: object

__init__(func, name='Periodic task', interval=3600, immediate_start=False, time_execution=False)

Run an arbitrary function func every interval seconds.

Set immediate_start to True to run func when task is started.

start()

run()

shutdown()

galaxy.util.template module

Entry point for the usage of Cheetah templating within Galaxy.

exception galaxy.util.template.InputNotFoundSyntaxError

Bases: SyntaxError

class galaxy.util.template.FixedModuleCodeCompiler(source=None, file=None, moduleName='DynamicallyCompiledCheetahTemplate', mainClassName=None, mainMethodName=None, baseclassName=None, extraImportStatements=None, settings=None)

Bases: ModuleCompiler

module_code = None

getModuleCode()

galaxy.util.template.create_compiler_class(module_code)

galaxy.util.template.fill_template(template_text, context=None, retry=10, compiler_class=<class 'Cheetah.Compiler.ModuleCompiler'>, first_exception=None, futurized=False, python_template_version='3', **kwargs)

Fill a cheetah template out for specified context.

If template_text is None, an exception will be thrown, if context is None (the default) - keyword arguments to this function will be used as the context.

galaxy.util.template.futurize_preprocessor(source)

galaxy.util.tool_version module

galaxy.util.tool_version.remove_version_from_guid(guid)

Removes version from toolshed-derived tool_id(=guid).

galaxy.util.topsort module

Topological sort.

From Tim Peters, see:

http://mail.python.org/pipermail/python-list/1999-July/006660.html

topsort takes a list of pairs, where each pair (x, y) is taken to mean that x <= y wrt some abstract partial ordering. The return value is a list, representing a total ordering that respects all the input constraints. E.g.,

topsort( [(1,2), (3,3)] )

Valid topological sorts would be any of (but nothing other than)

[3, 1, 2] [1, 3, 2] [1, 2, 3]

… however this variant ensures that ‘key’ order (first element of tuple) is preserved so the following will be result returned:

[1, 3, 2]

because those are the permutations of the input elements that respect the “1 precedes 2” and “3 precedes 3” input constraints. Note that a constraint of the form (x, x) is really just a trick to make sure x appears somewhere in the output list.

If there’s a cycle in the constraints, say

topsort( [(1,2), (2,1)] )

then CycleError is raised, and the exception object supports many methods to help analyze and break the cycles. This requires a good deal more code than topsort itself!

exception galaxy.util.topsort.CycleError(sofar, numpreds, succs)

Bases: Exception

__init__(sofar, numpreds, succs)

get_partial()

get_pred_counts()

get_succs()

get_elements()

get_pairlist()

get_preds()

pick_a_cycle()

galaxy.util.topsort.topsort(pairlist)

galaxy.util.topsort.topsort_levels(pairlist)

galaxy.util.ucsc module

Utilities for dealing with UCSC data.

exception galaxy.util.ucsc.UCSCLimitException

Bases: Exception

class galaxy.util.ucsc.UCSCOutWrapper(other)

Bases: object

File-like object that throws an exception if it encounters the UCSC limit error lines

__init__(other)

next()

readline()

galaxy.util.validation module

Module for validation of incoming inputs.

TODO: Refactor BaseController references to similar methods to use this module.

galaxy.util.validation.validate_and_sanitize_basestring(key, val)

galaxy.util.validation.validate_and_sanitize_basestring_list(key, val)

galaxy.util.validation.validate_boolean(key, val)

galaxy.util.watcher module

galaxy.util.watcher.get_observer_class(config_name, config_value, default, monitor_what_str)

galaxy.util.watcher.get_watcher(config, config_name, default='False', monitor_what_str=None, watcher_class=None, event_handler_class=None, **kwargs)

class galaxy.util.watcher.BaseWatcher(observer_class, event_handler_class, **kwargs)

Bases: object

__init__(observer_class, event_handler_class, **kwargs)

start()

monitor(dir_path, recursive=False)

resume_watching()

shutdown()

class galaxy.util.watcher.Watcher(observer_class, event_handler_class, **kwargs)

Bases: BaseWatcher

__init__(observer_class, event_handler_class, **kwargs)

watch_file(file_path, callback=None)

watch_directory(dir_path, callback=None, recursive=False, ignore_extensions=None, require_extensions=None)

class galaxy.util.watcher.EventHandler(watcher)

Bases: FileSystemEventHandler

__init__(watcher)

on_moved(event)

Called when a file or a directory is moved or renamed.

Parameters:

event (DirMovedEvent or FileMovedEvent) – Event representing file/directory movement.

on_created(event)

Called when a file or directory is created.

Parameters:

event (DirCreatedEvent or FileCreatedEvent) – Event representing file/directory creation.

on_deleted(event)

Called when a file or directory is deleted.

Parameters:

event (DirDeletedEvent or FileDeletedEvent) – Event representing file/directory deletion.

on_modified(event)

Called when a file or directory is modified.

Parameters:

event (DirModifiedEvent or FileModifiedEvent) – Event representing file/directory modification.

class galaxy.util.watcher.NullWatcher

Bases: object

start()

shutdown()

watch_file(*args, **kwargs)

watch_directory(*args, **kwargs)

galaxy.util.xml_macros module

galaxy.util.xml_macros.imported_macro_paths(root)

galaxy.util.xml_macros.load(path: str | PathLike) → ElementTree

galaxy.util.xml_macros.load_with_references(path: str | PathLike) → 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.util.xml_macros.raw_xml_tree(path: str | PathLike) → ElementTree

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

galaxy.util.xml_macros.template_macro_params(root)

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

galaxy.util.yaml_util module

class galaxy.util.yaml_util.OrderedLoader(stream)

Bases: CSafeLoader

__init__(stream)

include(node)

yaml_constructors = {'tag:yaml.org,2002:null': <function SafeConstructor.construct_yaml_null>, 'tag:yaml.org,2002:bool': <function SafeConstructor.construct_yaml_bool>, 'tag:yaml.org,2002:int': <function SafeConstructor.construct_yaml_int>, 'tag:yaml.org,2002:float': <function SafeConstructor.construct_yaml_float>, 'tag:yaml.org,2002:binary': <function SafeConstructor.construct_yaml_binary>, 'tag:yaml.org,2002:timestamp': <function SafeConstructor.construct_yaml_timestamp>, 'tag:yaml.org,2002:omap': <function SafeConstructor.construct_yaml_omap>, 'tag:yaml.org,2002:pairs': <function SafeConstructor.construct_yaml_pairs>, 'tag:yaml.org,2002:set': <function SafeConstructor.construct_yaml_set>, 'tag:yaml.org,2002:str': <function SafeConstructor.construct_yaml_str>, 'tag:yaml.org,2002:seq': <function SafeConstructor.construct_yaml_seq>, 'tag:yaml.org,2002:map': <function ordered_load.<locals>.construct_mapping>, None: <function SafeConstructor.construct_undefined>, '!include': <function OrderedLoader.include>}

galaxy.util.yaml_util.ordered_load(stream, merge_duplicate_keys=False)

Parse the first YAML document in a stream and produce the corresponding Python object.

If merge_duplicate_keys is True, merge the values of duplicate mapping keys into a list, as the uWSGI “dumb” YAML parser would do. Otherwise, following YAML 1.2 specification which says that “each key is unique in the association”, raise a ConstructionError exception.

galaxy.util.yaml_util.ordered_dump(data, stream=None, Dumper=<class 'yaml.dumper.Dumper'>, **kwds)

galaxy.util.zipstream module

class galaxy.util.zipstream.ZipstreamWrapper(archive_name: str | None = None, upstream_mod_zip: bool = False, upstream_gzip: bool = False)

Bases: object

__init__(archive_name: str | None = None, upstream_mod_zip: bool = False, upstream_gzip: bool = False) → None

response() → Iterator[bytes]

get_headers() → Dict[str, str]

add_path(path: str, archive_name: str) → None

write(path: str, archive_name: str | None = None) → None