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.managers package

Classes that manage resources (models, tools, etc.) by using the current Transaction.

Encapsulates the intersection of trans (or trans.sa_session), models, and Controllers.

Responsibilities:

  • model operations that involve the trans/sa_session (CRUD)
  • security: ownership, accessibility
  • common aspect-oriented operations via new mixins: sharable, annotatable, tagable, ratable

Not responsible for:

  • encoding/decoding ids
  • any http gobblygook
  • formatting of returned data (always python structures)
  • formatting of raised errors

The goal is to have Controllers only handle:

  • query-string/payload parsing and encoding/decoding ids
  • http
  • return formatting
  • control, improve namespacing in Controllers
  • DRY for Controller ops (define here - use in both UI/API Controllers)

In other words, ‘Business logic’ independent of web transactions/user context (trans) should be pushed into models - but logic that requires the context trans should be placed under this module.

Submodules

galaxy.managers.annotatable module

Mixins for Annotatable model managers and serializers.

class galaxy.managers.annotatable.AnnotatableDeserializerMixin[source]

Bases: object

add_deserializers()[source]
deserialize_annotation(item, key, val, user=None, **context)[source]

Make sure val is a valid annotation and assign it, deleting any existing if val is None.

class galaxy.managers.annotatable.AnnotatableFilterMixin[source]

Bases: object

filter_annotation_contains(item, val)[source]

Test whether val is in the owner’s annotation.

class galaxy.managers.annotatable.AnnotatableManagerMixin[source]

Bases: object

annotate(item, annotation, user=None, flush=True)[source]

Create a new annotation on item or delete the existing if annotation is None.

annotation(item)[source]

Return the annotation string made by the item’s owner or None if there is no annotation.

session() → galaxy.model.scoped_session.galaxy_scoped_session[source]
class galaxy.managers.annotatable.AnnotatableSerializerMixin[source]

Bases: object

add_serializers()[source]
serialize_annotation(item, key, user=None, **context)[source]

Get and serialize an item’s annotation.

galaxy.managers.api_keys module

class galaxy.managers.api_keys.ApiKeyManager(app: galaxy.structured_app.BasicSharedApp)[source]

Bases: object

__init__(app: galaxy.structured_app.BasicSharedApp)[source]
create_api_key(user) → str[source]
get_or_create_api_key(user) → str[source]

galaxy.managers.base module

Keeps the older BaseController security and fetching methods and also defines a base ModelManager, ModelSerializer, and ModelDeserializer.

ModelManagers are used for operations on models that occur outside the scope of a single model object, such as:

  • object creation
  • object lookup
  • interactions between 2+ objects of different model classes

(Since these were to replace model Mixins from web/framework/base/controller.py the rule of thumb used there also generally has been applied here: if it uses the trans or sa_session, put it in a manager and not the model.)

ModelSerializers allow flexible conversion of model objects to dictionaries. They control what keys are sent, how values are simplified, can remap keys, and allow both predefined and user controlled key sets.

ModelDeserializers control how a model validates and process an incoming attribute change to a model object.

class galaxy.managers.base.Deserializer(*args, **kwargs)[source]

Bases: typing.Protocol

__init__(*args, **kwargs)
class galaxy.managers.base.HasAModelManager(app: galaxy.structured_app.MinimalManagerApp, manager=None, **kwargs)[source]

Bases: typing.Generic

Mixin used where serializers, deserializers, filter parsers, etc. need some functionality around the model they’re mainly concerned with and would perform that functionality with a manager.

__init__(app: galaxy.structured_app.MinimalManagerApp, manager=None, **kwargs)[source]
manager

Return an appropriate manager if it exists, instantiate if not.

class galaxy.managers.base.ModelDeserializer(app: galaxy.structured_app.MinimalManagerApp, **kwargs)[source]

Bases: galaxy.managers.base.HasAModelManager

An object that converts an incoming serialized dict into values that can be directly assigned to an item’s attributes and assigns them.

__init__(app: galaxy.structured_app.MinimalManagerApp, **kwargs)[source]

Set up deserializers and validator.

add_deserializers()[source]

Register a map of attribute keys -> functions that will deserialize data into attributes to be assigned to the item.

default_deserializer(item, key, val, **context)[source]

If the incoming val is different than the item value change it and, in either case, return the value.

deserialize(item, data, flush=True, **context)[source]

Convert an incoming serialized dict into values that can be directly assigned to an item’s attributes and assign them

deserialize_basestring(item, key, val, convert_none_to_empty=False, **context)[source]
deserialize_bool(item, key, val, **context)[source]
deserialize_genome_build(item, key, val, **context)[source]

Make sure val is a valid dbkey and assign it.

deserialize_int(item, key, val, min=None, max=None, **context)[source]
validate = <galaxy.managers.base.ModelValidator object>
exception galaxy.managers.base.ModelDeserializingError(err_msg=None, type='info', **extra_error_info)[source]

Bases: galaxy.exceptions.ObjectAttributeInvalidException

Thrown when an incoming value isn’t usable by the model (bad type, out of range, etc.)

class galaxy.managers.base.ModelFilterParser(app: galaxy.structured_app.MinimalManagerApp, **kwargs)[source]

Bases: galaxy.managers.base.HasAModelManager

Converts string tuples (partially converted query string params) of attr, op, val into either:

  • ORM based filters (filters that can be applied by the ORM at the SQL level) or
  • functional filters (filters that use derived values or values not within the SQL tables)

These filters can then be applied to queries.

This abstraction allows ‘smarter’ application of limit and offset at either the SQL level or the generator/list level based on the presence of functional filters. In other words, if no functional filters are present, limit and offset may be applied at the SQL level. If functional filters are present, limit and offset need to applied at the list level.

These might be safely be replaced in the future by creating SQLAlchemy hybrid properties or more thoroughly mapping derived values.

UNDERSCORED_OPS = ('lt', 'le', 'eq', 'ne', 'ge', 'gt')
__init__(app: galaxy.structured_app.MinimalManagerApp, **kwargs)[source]

Set up serializer map, any additional serializable keys, and views here.

build_filter_params(query_params: galaxy.schema.FilterQueryParams, filter_attr_key: str = 'q', filter_value_key: str = 'qv', attr_op_split_char: str = '-') → typing.List[typing.Tuple[str, str, str]][source]

Builds a list of tuples containing filtering information in the form of (attribute, operator, value).

parse_date(date_string)[source]

Reformats a string containing either seconds from epoch or an iso8601 formated date string into a new date string usable within a filter query.

Seconds from epoch can be a floating point value as well (i.e containing ms).

parse_filter(attr, op, val)[source]

Attempt to parse filter as a custom/fn filter, then an orm filter, and if neither work - raise an error.

Raises:exceptions.RequestParameterInvalidException – if no functional or orm filter can be parsed.
parse_filters(filter_tuple_list)[source]

Parse string 3-tuples (attr, op, val) into orm or functional filters.

parse_id_list(id_list_string, sep=', ')[source]

Split id_list_string at sep.

parse_int_list(int_list_string, sep=', ')[source]

Split int_list_string at sep and parse as ints.

parse_query_filters(query_filters: galaxy.schema.FilterQueryParams)[source]

Convenience function to parse a FilterQueryParams object into a collection of filtering criteria.

parsed_filter

alias of ParsedFilter

string_standard_ops(key)[source]
class galaxy.managers.base.ModelManager(app: galaxy.structured_app.BasicSharedApp)[source]

Bases: object

Base class for all model/resource managers.

Provides common queries and CRUD operations as a (hopefully) light layer over the ORM.

__init__(app: galaxy.structured_app.BasicSharedApp)[source]
associate(associate_with, item, foreign_key_name=None)[source]

Generically associate item with associate_with based on foreign_key_name.

by_id(id: int)[source]

Gets a model by primary id.

by_ids(ids, filters=None, **kwargs)[source]

Returns an in-order list of models with the matching ids in ids.

copy(item, **kwargs)[source]

Clone or copy an item.

create(flush=True, *args, **kwargs)[source]

Generically create a new model.

list(filters=None, order_by=None, limit=None, offset=None, **kwargs)[source]

Returns all objects matching the given filters

one(**kwargs)[source]

Sends kwargs to build the query and returns one and only one model.

query(eagerloads: bool = True, filters=None, order_by=None, limit: typing.Optional[int] = None, offset: typing.Optional[int] = None) → sqlalchemy.orm.query.Query[source]

Return a basic query from model_class, filters, order_by, and limit and offset.

Set eagerloads to False to disable them for this query.

query_associated(associated_model_class, item, foreign_key_name=None)[source]

Generically query other items that have been associated with this item.

session() → sqlalchemy.orm.scoping.scoped_session[source]
update(item, new_values, flush=True, **kwargs)[source]

Given a dictionary of new values, update item and return it.

..note: NO validation or deserialization occurs here.

class galaxy.managers.base.ModelSerializer(app: galaxy.structured_app.MinimalManagerApp, **kwargs)[source]

Bases: galaxy.managers.base.HasAModelManager

Turns models into JSONable dicts.

Maintains a map of requestable keys and the Callable() serializer functions that should be called for those keys. E.g. { ‘x’ : lambda item, key: item.x, … }

Note: if a key to serialize is not listed in the Serializer.serializable_keyset or serializers, it will not be returned.

To serialize call:
my_serializer = MySerializer( app ) … keys_to_serialize = [ ‘id’, ‘name’, ‘attr1’, ‘attr2’, … ] item_dict = MySerializer.serialize( my_item, keys_to_serialize )
__init__(app: galaxy.structured_app.MinimalManagerApp, **kwargs)[source]

Set up serializer map, any additional serializable keys, and views here.

add_serializers()[source]

Register a map of attribute keys -> serializing functions that will serialize the attribute.

add_view(view_name, key_list, include_keys_from=None)[source]

Add the list of serializable attributes key_list to the serializer’s view dictionary under the key view_name.

If include_keys_from is a proper view name, extend key_list by the list in that view.

default_serializer(item, key, **context)[source]

Serialize the item’s attribute named key.

serialize(item, keys, **context)[source]

Serialize the model item to a dictionary.

Given model item and the list keys, create and return a dictionary built from each key in keys that also exists in serializers and values of calling the keyed/named serializers on item.

serialize_date(item: typing.Any, key: str, **context)[source]

Serialize a date attribute of item.

serialize_id(item: typing.Any, key: str, **context)[source]

Serialize an id attribute of item.

serialize_to_view(item, view=None, keys=None, default_view=None, **context)[source]

Use a predefined list of keys (the string view) and any additional keys listed in keys.

The combinations can be:
view only: return those keys listed in the named view keys only: return those keys listed no view or keys: use the default_view if any view and keys: combine both into one list of keys
serialize_type_id(item: typing.Any, key: str, **context)[source]

Serialize an type-id for item.

skip(msg='skipped')[source]

To be called from inside a serializer to skip it.

Handy for config checks, information hiding, etc.

static url_for(*args, context=None, **kwargs)[source]
exception galaxy.managers.base.ModelSerializingError(err_msg=None, type='info', **extra_error_info)[source]

Bases: galaxy.exceptions.InternalServerError

Thrown when request model values can’t be serialized

class galaxy.managers.base.ModelValidator[source]

Bases: object

An object that inspects a dictionary (generally meant to be a set of new/updated values for the model) and raises an error if a value is not acceptable.

static basestring(key: str, val: typing.Any) → str[source]
static basestring_list(key: str, val: typing.Any) → typing.List[str][source]

Must be a list of basestrings.

static bool(key: str, val: typing.Any) → bool[source]
static genome_build(key: str, val: typing.Any) → str[source]

Must be a valid base_string.

Note: no checking against installation’s ref list is done as many data sources consider this an open field.

static int_range(key: str, val: typing.Any, min: typing.Optional[int] = None, max: typing.Optional[int] = None) → int[source]

Must be a int between min and max.

static matches_type(key: str, val: typing.Any, types: typing.Union[type, typing.Tuple[typing.Union[type, typing.Tuple[typing.Any, ...]], ...]])[source]

Check val against the type (or tuple of types) in types.

Raises:exceptions.RequestParameterInvalidException – if not an instance.
static nullable_basestring(key: str, val: typing.Any) → str[source]

Must be a basestring or None.

class galaxy.managers.base.ParsedFilter(filter_type, filter)[source]

Bases: tuple

filter

Alias for field number 1

filter_type

Alias for field number 0

class galaxy.managers.base.Serializer(*args, **kwargs)[source]

Bases: typing.Protocol

__init__(*args, **kwargs)
exception galaxy.managers.base.SkipAttribute[source]

Bases: Exception

Raise this inside a serializer to prevent the returned dictionary from having a the associated key or value for this attribute.

class galaxy.managers.base.SortableManager[source]

Bases: object

A manager interface for parsing order_by strings into actual ‘order by’ queries.

parse_order_by(order_by_string, default=None)[source]

Return an ORM compatible order_by clause using the given string (i.e.: ‘name-dsc,create_time’). This must be implemented by the manager.

galaxy.managers.base.decode_id(app: galaxy.structured_app.BasicSharedApp, id: typing.Any)[source]
galaxy.managers.base.decode_with_security(security: galaxy.security.idencoding.IdEncodingHelper, id: typing.Any)[source]
galaxy.managers.base.encode_with_security(security: galaxy.security.idencoding.IdEncodingHelper, id: typing.Any)[source]
galaxy.managers.base.get_class(class_name)[source]

Returns the class object that a string denotes. Without this method, we’d have to do eval(<class_name>).

galaxy.managers.base.get_object(trans, id, class_name, check_ownership=False, check_accessible=False, deleted=None)[source]

Convenience method to get a model object with the specified checks. This is a generic method for dealing with objects uniformly from the older controller mixin code - however whenever possible the managers for a particular model should be used to load objects.

galaxy.managers.base.is_valid_slug(slug)[source]

Returns true iff slug is valid.

galaxy.managers.base.munge_lists(listA, listB)[source]

Combine two lists into a single list.

(While allowing them to be None, non-lists, or lists.)

galaxy.managers.base.parse_bool(bool_string: typing.Union[str, bool]) → bool[source]

Parse a boolean from a string.

galaxy.managers.base.parsed_filter

alias of galaxy.managers.base.ParsedFilter

galaxy.managers.base.raise_filter_err(attr, op, val, msg)[source]
galaxy.managers.base.security_check(trans, item, check_ownership=False, check_accessible=False)[source]

Security checks for an item: checks if (a) user owns item or (b) item is accessible to user. This is a generic method for dealing with objects uniformly from the older controller mixin code - however whenever possible the managers for a particular model should be used to perform security checks.

galaxy.managers.citations module

class galaxy.managers.citations.BaseCitation[source]

Bases: object

equals(other_citation)[source]
has_doi()[source]
to_dict(citation_format)[source]
class galaxy.managers.citations.BibtexCitation(elem, citation_manager)[source]

Bases: galaxy.managers.citations.BaseCitation

__init__(elem, citation_manager)[source]
to_bibtex()[source]
class galaxy.managers.citations.CitationCollection[source]

Bases: object

__init__()[source]
add(new_citation)[source]
class galaxy.managers.citations.CitationsManager(app: galaxy.structured_app.BasicSharedApp) → None[source]

Bases: object

__init__(app: galaxy.structured_app.BasicSharedApp) → None[source]
citations_for_tool(tool)[source]
citations_for_tool_ids(tool_ids)[source]
parse_citation(citation_elem)[source]
class galaxy.managers.citations.DoiCache(config)[source]

Bases: object

__init__(config)[source]
get_bibtex(doi)[source]
class galaxy.managers.citations.DoiCitation(elem, citation_manager)[source]

Bases: galaxy.managers.citations.BaseCitation

BIBTEX_UNSET = <object object>
__init__(elem, citation_manager)[source]
doi()[source]
has_doi()[source]
to_bibtex()[source]
galaxy.managers.citations.parse_citation(elem, citation_manager)[source]

Parse an abstract citation entry from the specified XML element.

galaxy.managers.cloud module

Manager and serializer for cloud-based storages.

class galaxy.managers.cloud.CloudManager(app: galaxy.structured_app.MinimalManagerApp)[source]

Bases: galaxy.managers.sharable.SharableModelManager

static configure_provider(provider, credentials)[source]

Given a provider name and required credentials, it configures and returns a cloudbridge connection to the provider.

Parameters:
  • provider (string) – the name of cloud-based resource provided. A list of supported providers is given in SUPPORTED_PROVIDERS variable.
  • credentials (dict) – a dictionary containing all the credentials required to authenticated to the specified provider.
Return type:

provider specific, e.g., cloudbridge.cloud.providers.aws.provider.AWSCloudProvider for AWS.

Returns:

a cloudbridge connection to the specified provider.

get(trans, history_id, bucket_name, objects, authz_id, input_args=None)[source]

Implements the logic of getting a file from a cloud-based storage (e.g., Amazon S3) and persisting it as a Galaxy dataset.

This manager does NOT require use credentials, instead, it uses a more secure method, which leverages CloudAuthz (https://github.com/galaxyproject/cloudauthz) and automatically requests temporary credentials to access the defined resources.

Parameters:
  • trans (galaxy.webapps.base.webapp.GalaxyWebTransaction) – Galaxy web transaction
  • history_id (string) – the (decoded) id of history to which the object should be received to.
  • bucket_name (string) – the name of a bucket from which data should be fetched (e.g., a bucket name on AWS S3).
  • objects (list of string) – the name of objects to be fetched.
  • authz_id (int) – the ID of CloudAuthz to be used for authorizing access to the resource provider. You may get a list of the defined authorizations sending GET to /api/cloud/authz. Also, you can POST to /api/cloud/authz to define a new authorization.
  • input_args (dict) – a [Optional] a dictionary of input parameters: dbkey, file_type, space_to_tab, to_posix_lines (see galaxy/webapps/galaxy/api/cloud.py)
Return type:

list of galaxy.model.Dataset

Returns:

a list of datasets created for the fetched files.

model_class

alias of galaxy.model.History

send(trans, history_id, bucket_name, authz_id, dataset_ids=None, overwrite_existing=False)[source]

Implements the logic of sending dataset(s) from a given history to a given cloud-based storage (e.g., Amazon S3).

Parameters:
  • trans (galaxy.webapps.base.webapp.GalaxyWebTransaction) – Galaxy web transaction
  • history_id (string) – the (encoded) id of history from which the object should be sent.
  • bucket_name (string) – the name of a bucket to which data should be sent (e.g., a bucket name on AWS S3).
  • authz_id (int) – the ID of CloudAuthz to be used for authorizing access to the resource provider. You may get a list of the defined authorizations via /api/cloud/authz. Also, you can use /api/cloud/authz/create to define a new authorization.
  • dataset_ids (set) – [Optional] The list of (decoded) dataset ID(s) belonging to the given history which should be sent to the given provider. If not provided, Galaxy sends all the datasets belonging to the given history.
  • overwrite_existing (boolean) – [Optional] If set to “True”, and an object with same name of the dataset to be sent already exist in the bucket, Galaxy replaces the existing object with the dataset to be sent. If set to “False”, Galaxy appends datetime to the dataset name to prevent overwriting the existing object.
Return type:

tuple

Returns:

A tuple of two lists of labels of the objects that were successfully and unsuccessfully sent to cloud.

galaxy.managers.cloudauthzs module

Manager and (de)serializer for cloud authorizations (cloudauthzs).

class galaxy.managers.cloudauthzs.CloudAuthzManager(app: galaxy.structured_app.MinimalManagerApp)[source]

Bases: galaxy.managers.sharable.SharableModelManager

foreign_key_name = 'cloudauthz'
model_class

alias of galaxy.model.CloudAuthz

class galaxy.managers.cloudauthzs.CloudAuthzsDeserializer(app: galaxy.structured_app.MinimalManagerApp, **kwargs)[source]

Bases: galaxy.managers.base.ModelDeserializer

Service object for validating and deserializing dictionaries that update/alter cloudauthz configurations.

add_deserializers()[source]
deserialize_and_validate_authn_id(item, key, val, **context)[source]

Deserializes an authentication ID (authn_id), and asserts if the current user can assume that authentication.

Parameters:
  • item (galaxy.model.CloudAuthz) – an instance of cloudauthz
  • key (string) – authn_id attribute of the cloudauthz object (i.e., the item param).
  • val (string) – the value of authn_id attribute of the cloudauthz object (i.e., the item param).
  • context (dict) – a dictionary object containing Galaxy trans.
Return type:

string

Returns:

decoded authentication ID.

model_manager_class

alias of CloudAuthzManager

class galaxy.managers.cloudauthzs.CloudAuthzsSerializer(app, **kwargs)[source]

Bases: galaxy.managers.base.ModelSerializer

Interface/service object for serializing cloud authorizations (cloudauthzs) into dictionaries.

__init__(app, **kwargs)[source]
add_serializers()[source]
model_manager_class

alias of CloudAuthzManager

galaxy.managers.collections module

class galaxy.managers.collections.DatasetCollectionManager(model: galaxy.model.mapping.GalaxyModelMapping, security: galaxy.security.idencoding.IdEncodingHelper, hda_manager: galaxy.managers.hdas.HDAManager, history_manager: galaxy.managers.histories.HistoryManager, tag_handler: galaxy.model.tags.GalaxyTagHandler, ldda_manager: galaxy.managers.lddas.LDDAManager)[source]

Bases: object

Abstraction for interfacing with dataset collections instance - ideally abstracts out model and plugin details.

ELEMENTS_UNINITIALIZED = <object object>
__init__(model: galaxy.model.mapping.GalaxyModelMapping, security: galaxy.security.idencoding.IdEncodingHelper, hda_manager: galaxy.managers.hdas.HDAManager, history_manager: galaxy.managers.histories.HistoryManager, tag_handler: galaxy.model.tags.GalaxyTagHandler, ldda_manager: galaxy.managers.lddas.LDDAManager)[source]
apply_rules(hdca, rule_set, handle_dataset)[source]
collection_builder_for(dataset_collection)[source]
copy(trans, parent, source, encoded_source_id, copy_elements=False, dataset_instance_attributes=None)[source]

PRECONDITION: security checks on ability to add to parent occurred during load.

create(trans, parent, name, collection_type, element_identifiers=None, elements=None, implicit_collection_info=None, trusted_identifiers=None, hide_source_items=False, tags=None, copy_elements=False, history=None, set_hid=True, flush=True, completed_job=None, output_name=None)[source]

PRECONDITION: security checks on ability to add to parent occurred during load.

create_dataset_collection(trans, collection_type, element_identifiers=None, elements=None, hide_source_items=None, copy_elements=False, history=None)[source]
delete(trans, instance_type, id, recursive=False, purge=False)[source]
get_collection_contents(trans, parent_id, limit=None, offset=None)[source]

Find first level of collection contents by containing collection parent_id

get_converters_for_collection(trans, id, datatypes_registry: galaxy.datatypes.registry.Registry, instance_type='history')[source]
get_dataset_collection(trans, encoded_id)[source]
get_dataset_collection_instance(trans, instance_type, id, **kwds)[source]
history_dataset_collections(history, query)[source]
match_collections(collections_to_match)[source]

May seem odd to place it here, but planning to grow sophistication and get plugin types involved so it will likely make sense in the future.

precreate_dataset_collection(structure, allow_unitialized_element=True, completed_collection=None, implicit_output_name=None)[source]
precreate_dataset_collection_instance(trans, parent, name, structure, implicit_inputs=None, implicit_output_name=None, tags=None, completed_collection=None)[source]
update(trans, instance_type, id, payload)[source]

galaxy.managers.collections_util module

galaxy.managers.collections_util.api_payload_to_create_params(payload)[source]

Cleanup API payload to pass into dataset_collections.

galaxy.managers.collections_util.dictify_dataset_collection_instance(dataset_collection_instance, parent, security, url_builder, view='element', fuzzy_count=None)[source]

galaxy.managers.configuration module

Managers, serializers for Galaxy config file data. ConfigSerializer for all users and a more expanded set of data for admin in AdminConfigSerializer.

Used by both the API and bootstrapped data.

class galaxy.managers.configuration.AdminConfigSerializer(app)[source]

Bases: galaxy.managers.configuration.ConfigSerializer

Configuration attributes viewable only by admin users

add_serializers()[source]
class galaxy.managers.configuration.ConfigSerializer(app)[source]

Bases: galaxy.managers.base.ModelSerializer

Configuration (galaxy.ini) settings viewable by all users

__init__(app)[source]
add_serializers()[source]
default_serializer(config, key)[source]
class galaxy.managers.configuration.ConfigurationManager(app: galaxy.structured_app.StructuredApp)[source]

Bases: object

Interface/service object for interacting with configuration and related data.

__init__(app: galaxy.structured_app.StructuredApp)[source]
decode_id(encoded_id: galaxy.schema.fields.EncodedDatabaseIdField) → typing.Dict[str, int][source]
dynamic_tool_confs() → typing.List[typing.Dict[str, str]][source]
get_configuration(trans: galaxy.managers.context.ProvidesUserContext, serialization_params: galaxy.schema.SerializationParams) → typing.Dict[str, typing.Any][source]
reload_toolbox()[source]
tool_lineages() → typing.List[typing.Dict[str, typing.Dict]][source]
version() → typing.Dict[str, typing.Any][source]

galaxy.managers.containers module

galaxy.managers.context module

Interfaces/mixins for transaction-like objects.

These objects describe the context around a unit of work. This unit of work is very broad and can be anything from the response to a web request, the scheduling of a workflow, the reloading the toolbox, etc.. Traditionally, Galaxy has simply passed around a GalaxyWebTransaction object through all layers and large components of the Galaxy app. Having random backend components define explicit dependencies on this however is inappropriate because Galaxy may not be used in all sort of non-web contexts. The future use of message queues and web sockets as well as the decomposition of the backend into packages only further make this heavy reliance on GalaxyWebTransaction inappropriate.

A better approach is for components to annotate their reliance on much narrower, typed views of the GalaxyWebTransaction. This allows explicit declaration of what is being required in the context of a method or class and allows the Python type system to ensure the transaction supplied to the method is appropriate for the context. For instance, an effective use of the type system in this way can prevent the backend work context used to schedule workflow from being supplied to a method that requires an older-style WSGI web response object.

There are various levels of transactions defined in this file - these include galaxy.managers.context.ProvidesAppContext, galaxy.managers.context.ProvidesUserContext, and galaxy.managers.context.ProvidesHistoryContext. Methods should annotate their dependency on the narrowest context they require. A method that requires a user but not a history should declare its trans argument as requiring type galaxy.managers.context.ProvidesUserContext.

class galaxy.managers.context.ProvidesAppContext[source]

Bases: object

For transaction-like objects to provide Galaxy convenience layer for database and event handling.

Mixed in class must provide app property.

app

Provide access to the Galaxy app object.

url_builder

Provide access to Galaxy URLs (if available).

Parameters:qualified (bool) – if True, the fully qualified URL is returned, else a relative URL is returned (default False).
security

Provide access to Galaxy app’s id encoding helper.

Return type:IdEncodingHelper
log_action(user=None, action=None, context=None, params=None)[source]

Application-level logging of user actions.

log_event(message, tool_id=None, **kwargs)[source]

Application level logging. Still needs fleshing out (log levels and such) Logging events is a config setting - if False, do not log.

sa_session

Provide access to Galaxy’s SQLAlchemy session.

Return type:galaxy.model.scoped_session.galaxy_scoped_session
expunge_all()[source]

Expunge all the objects in Galaxy’s SQLAlchemy sessions.

get_toolbox()[source]

Returns the application toolbox.

Return type:galaxy.tools.ToolBox
model

Provide access to Galaxy’s model mapping class.

This is sometimes used for quick access to classes in galaxy.model but this is discouraged. Those classes should be imported by the consumer for stronger static checking.

This is more proper use for this is accessing the threadbound SQLAlchemy session or engine.

Return type:galaxy.model.base.ModelMapping
install_model

Provide access to Galaxy’s install mapping.

Comments on the model property apply here also.

class galaxy.managers.context.ProvidesUserContext[source]

Bases: galaxy.managers.context.ProvidesAppContext

For transaction-like objects to provide Galaxy convenience layer for reasoning about users.

Mixed in class must provide user and app properties.

user

Provide access to the user object.

user_vault

Provide access to a user’s personal vault.

anonymous
get_current_user_roles() → typing.List[galaxy.model.Role][source]
user_is_admin
user_is_bootstrap_admin

Master key provided so there is no real user

user_can_do_run_as
user_is_active
check_user_activation()[source]

If user activation is enabled and the user is not activated throw an exception.

user_ftp_dir
class galaxy.managers.context.ProvidesHistoryContext[source]

Bases: galaxy.managers.context.ProvidesUserContext

For transaction-like objects to provide Galaxy convenience layer for reasoning about histories.

Mixed in class must provide user, history, and app properties.

history

Provide access to the user’s current history model object.

Return type:Optional[galaxy.model.History]
db_dataset_for(dbkey) → typing.Optional[galaxy.model.HistoryDatasetAssociation][source]

Optionally return the db_file dataset associated/needed by dataset.

db_builds

Returns the builds defined by galaxy and the builds defined by the user (chromInfo in history).

galaxy.managers.datasets module

Manager and Serializer for Datasets.

class galaxy.managers.datasets.DatasetAssociationDeserializer(app: galaxy.structured_app.MinimalManagerApp, **kwargs)[source]

Bases: galaxy.managers.base.ModelDeserializer, galaxy.managers.deletable.PurgableDeserializerMixin

add_deserializers()[source]
deserialize_datatype(item, key, val, **context)[source]
deserialize_metadata(dataset_assoc, metadata_key, metadata_dict, **context)[source]
deserialize_metadatum(dataset_assoc, key, val, **context)[source]
class galaxy.managers.datasets.DatasetAssociationFilterParser(app: galaxy.structured_app.MinimalManagerApp, **kwargs)[source]

Bases: galaxy.managers.base.ModelFilterParser, galaxy.managers.deletable.PurgableFiltersMixin

eq_datatype(dataset_assoc, class_str)[source]

Is the dataset_assoc datatype equal to the registered datatype class_str?

isinstance_datatype(dataset_assoc, class_strs)[source]

Is the dataset_assoc datatype derived from any of the registered datatypes in the comma separated string class_strs?

class galaxy.managers.datasets.DatasetAssociationManager(app)[source]

Bases: galaxy.managers.base.ModelManager, galaxy.managers.secured.AccessibleManagerMixin, galaxy.managers.deletable.PurgableManagerMixin

DatasetAssociation/DatasetInstances are intended to be working proxies to a Dataset, associated with either a library or a user/history (HistoryDatasetAssociation).

__init__(app)[source]
by_user(user)[source]
creating_job(dataset_assoc)[source]

Return the Job that created this dataset or None if not found.

detect_datatype(trans, dataset_assoc)[source]

Sniff and assign the datatype to a given dataset association (ldda or hda)

extra_files(dataset_assoc)[source]

Return a list of file paths for composite files, an empty list otherwise.

is_accessible(dataset_assoc, user, **kwargs)[source]

Is this DA accessible to user?

is_composite(dataset_assoc)[source]

Return True if this hda/ldda is a composite type dataset.

Note

see also (whereever we keep information on composite datatypes?)

purge(dataset_assoc, flush=True)[source]

Purge this DatasetInstance and the dataset underlying it.

serialize_dataset_association_roles(trans, dataset_assoc)[source]
set_metadata(trans, dataset_assoc, overwrite=False, validate=True)[source]

Trigger a job that detects and sets metadata on a given dataset association (ldda or hda)

stop_creating_job(dataset_assoc)[source]

Stops an dataset_assoc’s creating job if all the job’s other outputs are deleted.

update_permissions(trans, dataset_assoc, **kwd)[source]
class galaxy.managers.datasets.DatasetAssociationSerializer(app)[source]

Bases: galaxy.managers.datasets._UnflattenedMetadataDatasetAssociationSerializer

add_serializers()[source]
serialize(dataset_assoc, keys, **context)[source]

Override to add metadata as flattened keys on the serialized DatasetInstance.

class galaxy.managers.datasets.DatasetManager(app: galaxy.structured_app.MinimalManagerApp)[source]

Bases: galaxy.managers.base.ModelManager, galaxy.managers.secured.AccessibleManagerMixin, galaxy.managers.deletable.PurgableManagerMixin

Manipulate datasets: the components contained in DatasetAssociations/DatasetInstances/HDAs/LDDAs

__init__(app: galaxy.structured_app.MinimalManagerApp)[source]
copy(dataset, **kwargs)[source]
create(manage_roles=None, access_roles=None, flush=True, **kwargs)[source]

Create and return a new Dataset object.

error_unless_dataset_purge_allowed(msg=None)[source]
foreign_key_name = 'dataset'
has_access_permission(dataset, user)[source]

Return T/F if the user has role-based access to the dataset.

is_accessible(dataset, user, **kwargs)[source]

Is this dataset readable/viewable to user?

model_class

alias of galaxy.model.Dataset

purge(dataset, flush=True)[source]

Remove the object_store/file for this dataset from storage and mark as purged.

Raises:exceptions.ConfigDoesNotAllowException – if the instance doesn’t allow
class galaxy.managers.datasets.DatasetRBACPermissions(app)[source]

Bases: object

__init__(app)[source]
available_roles(trans, dataset, controller='root')[source]
get(dataset, flush=True)[source]
set(dataset, manage_roles, access_roles, flush=True)[source]
set_private_to_one_user(dataset, user, flush=True)[source]
set_public_with_single_manager(dataset, user, flush=True)[source]
class galaxy.managers.datasets.DatasetSerializer(app: galaxy.structured_app.MinimalManagerApp, user_manager: galaxy.managers.users.UserManager)[source]

Bases: galaxy.managers.base.ModelSerializer, galaxy.managers.deletable.PurgableSerializerMixin

__init__(app: galaxy.structured_app.MinimalManagerApp, user_manager: galaxy.managers.users.UserManager)[source]
add_serializers()[source]
model_manager_class

alias of DatasetManager

serialize_extra_files_path(item, key, user=None, **context)[source]

If the config allows or the user is admin, return the file path.

serialize_file_name(item, key, user=None, **context)[source]

If the config allows or the user is admin, return the file name of the file that contains this dataset’s data.

serialize_permissions(item, key, user=None, **context)[source]

galaxy.managers.deletable module

Many models in Galaxy are not meant to be removed from the database but only marked as deleted. These models have the boolean attribute ‘deleted’.

Other models are deletable and also may be purged. Most often these are models have some backing/supporting resources that can be removed as well (e.g. Datasets have data files on a drive). Purging these models removes the supporting resources as well. These models also have the boolean attribute ‘purged’.

class galaxy.managers.deletable.DeletableDeserializerMixin[source]

Bases: object

add_deserializers()[source]
deserialize_deleted(item, key, val, **context)[source]

Delete or undelete item based on val then return item.deleted.

class galaxy.managers.deletable.DeletableFiltersMixin[source]

Bases: object

class galaxy.managers.deletable.DeletableManagerMixin[source]

Bases: object

A mixin/interface for a model that is deletable (i.e. has a ‘deleted’ attr).

Many resources in Galaxy can be marked as deleted - meaning (in most cases) that they are no longer needed, should not be displayed, or may be actually removed by an admin/script.

delete(item, flush=True, **kwargs)[source]

Mark as deleted and return.

undelete(item, flush=True, **kwargs)[source]

Mark as not deleted and return.

class galaxy.managers.deletable.DeletableSerializerMixin[source]

Bases: object

add_serializers()[source]
class galaxy.managers.deletable.PurgableDeserializerMixin[source]

Bases: galaxy.managers.deletable.DeletableDeserializerMixin

add_deserializers()[source]
deserialize_purged(item, key, val, **context)[source]

If val is True, purge item and return item.purged.

deserializers = {}
class galaxy.managers.deletable.PurgableFiltersMixin[source]

Bases: galaxy.managers.deletable.DeletableFiltersMixin

class galaxy.managers.deletable.PurgableManagerMixin[source]

Bases: galaxy.managers.deletable.DeletableManagerMixin

A manager interface/mixin for a resource that allows deleting and purging where purging is often removal of some additional, non-db resource (e.g. a dataset’s file).

purge(item, flush=True, **kwargs)[source]

Mark as purged and return.

Override this in subclasses to do the additional resource removal.

class galaxy.managers.deletable.PurgableSerializerMixin[source]

Bases: galaxy.managers.deletable.DeletableSerializerMixin

add_serializers()[source]

galaxy.managers.executables module

Utilities for loading tools and workflows from paths for admin user requests.

galaxy.managers.executables.artifact_class(trans, as_dict)[source]

galaxy.managers.folders module

Manager and Serializer for Library Folders.

class galaxy.managers.folders.FolderManager[source]

Bases: object

Interface/service object for interacting with folders.

can_add_item(trans, folder)[source]

Return true if the user has permissions to add item to the given folder.

check_accessible(trans, folder)[source]

Check whether the folder is accessible to current user. By default every folder is accessible (contents have their own permissions).

check_manageable(trans, folder)[source]

Check whether the user can manage the folder.

Returns:the original folder
Return type:LibraryFolder
Raises:AuthenticationRequired, InsufficientPermissionsException
check_modifyable(trans, folder)[source]

Check whether the user can modify the folder (name and description).

Returns:the original folder
Return type:LibraryFolder
Raises:AuthenticationRequired, InsufficientPermissionsException
create(trans, parent_folder_id, new_folder_name, new_folder_description='')[source]

Create a new folder under the given folder.

Parameters:
  • parent_folder_id (int) – decoded id
  • new_folder_name (str) – name of the new folder
  • new_folder_description (str) – description of the folder (optional, defaults to empty string)
Returns:

the new folder

Return type:

LibraryFolder

Raises:

InsufficientPermissionsException

cut_and_decode(trans, encoded_folder_id)[source]

Cuts the folder prefix (the prepended ‘F’) and returns the decoded id.

Parameters:encoded_folder_id (string) – encoded id of the Folder object
Returns:decoded Folder id
Return type:int
cut_the_prefix(encoded_folder_id)[source]

Remove the prefix from the encoded folder id.

Parameters:encoded_folder_id (string) – encoded id of the Folder object with ‘F’ prepended
Returns:encoded Folder id without the ‘F’ prefix
Return type:string
Raises:MalformedId
decode_folder_id(trans, encoded_folder_id)[source]

Decode the folder id given that it has already lost the prefixed ‘F’.

Parameters:encoded_folder_id (string) – encoded id of the Folder object
Returns:decoded Folder id
Return type:int
Raises:MalformedId
delete(trans, folder, undelete=False)[source]

Mark given folder deleted/undeleted based on the flag.

Parameters:
  • folder (LibraryFolder) – the model object
  • undelete (Bool) – flag whether to delete (when False) or undelete
Returns:

the folder

Return type:

LibraryFolder

Raises:

ItemAccessibilityException

get(trans, decoded_folder_id, check_manageable=False, check_accessible=True)[source]

Get the folder from the DB.

Parameters:
  • decoded_folder_id (int) – decoded folder id
  • check_manageable (bool) – flag whether the check that user can manage item
  • check_accessible (bool) – flag whether to check that user can access item
Returns:

the requested folder

Return type:

LibraryFolder

Raises:

InconsistentDatabase, RequestParameterInvalidException, InternalServerError

get_current_roles(trans, folder)[source]

Find all roles currently connected to relevant permissions on the folder.

Parameters:folder (LibraryFolder) – the model object
Returns:dict of current roles for all available permission types
Return type:dictionary
get_folder_dict(trans, folder)[source]

Return folder data in the form of a dictionary.

Parameters:folder (LibraryFolder) – folder item
Returns:dict with data about the folder
Return type:dictionary
secure(trans, folder, check_manageable=True, check_accessible=True)[source]

Check if (a) user can manage folder or (b) folder is accessible to user.

Parameters:
  • folder (LibraryFolder) – folder item
  • check_manageable (bool) – flag whether to check that user can manage item
  • check_accessible (bool) – flag whether to check that user can access item
Returns:

the original folder

Return type:

LibraryFolder

update(trans, folder, name=None, description=None)[source]

Update the given folder’s name or description.

Parameters:
  • folder (LibraryFolder) – the model object
  • name (str) – new name for the library folder
  • description (str) – new description for the library folder
Returns:

the folder

Return type:

LibraryFolder

Raises:

ItemAccessibilityException, InsufficientPermissionsException

galaxy.managers.hdas module

Manager and Serializer for HDAs.

HistoryDatasetAssociations (HDAs) are datasets contained or created in a history.

class galaxy.managers.hdas.HDADeserializer(app: galaxy.structured_app.MinimalManagerApp)[source]

Bases: galaxy.managers.datasets.DatasetAssociationDeserializer, galaxy.managers.taggable.TaggableDeserializerMixin, galaxy.managers.annotatable.AnnotatableDeserializerMixin

Interface/service object for validating and deserializing dictionaries into histories.

__init__(app: galaxy.structured_app.MinimalManagerApp)[source]
add_deserializers()[source]
model_manager_class

alias of HDAManager

class galaxy.managers.hdas.HDAFilterParser(app: galaxy.structured_app.MinimalManagerApp, **kwargs)[source]

Bases: galaxy.managers.datasets.DatasetAssociationFilterParser, galaxy.managers.taggable.TaggableFilterMixin, galaxy.managers.annotatable.AnnotatableFilterMixin

model_class

alias of galaxy.model.HistoryDatasetAssociation

model_manager_class

alias of HDAManager

class galaxy.managers.hdas.HDAManager(app: galaxy.structured_app.MinimalManagerApp, user_manager: galaxy.managers.users.UserManager, tag_handler: galaxy.model.tags.GalaxyTagHandler)[source]

Bases: galaxy.managers.datasets.DatasetAssociationManager, galaxy.managers.secured.OwnableManagerMixin, galaxy.managers.taggable.TaggableManagerMixin, galaxy.managers.annotatable.AnnotatableManagerMixin

Interface/service object for interacting with HDAs.

__init__(app: galaxy.structured_app.MinimalManagerApp, user_manager: galaxy.managers.users.UserManager, tag_handler: galaxy.model.tags.GalaxyTagHandler)[source]

Set up and initialize other managers needed by hdas.

annotation(hda)[source]
annotation_assoc

alias of galaxy.model.HistoryDatasetAssociationAnnotationAssociation

copy(hda, history=None, hide_copy=False, flush=True, **kwargs)[source]

Copy hda, including annotation and tags, add to history and return the given HDA.

copy_ldda(history, ldda, **kwargs)[source]

Copy this HDA as a LDDA and return.

create(history=None, dataset=None, flush=True, **kwargs)[source]

Create a new hda optionally passing in it’s history and dataset.

..note: to explicitly set hid to None you must pass in hid=None, otherwise it will be automatically set.

data_conversion_status(hda)[source]

Returns a message if an hda is not ready to be used in visualization.

error_if_uploading(hda)[source]

Raise error if HDA is still uploading.

foreign_key_name = 'history_dataset_association'
get_owned_ids(object_ids, history=None)[source]

Get owned IDs.

has_been_resubmitted(hda)[source]

Return True if the hda’s job was resubmitted at any point.

is_accessible(hda, user, **kwargs)[source]

Override to allow owners (those that own the associated history).

is_owner(hda, user, current_history=None, **kwargs)[source]

Use history to see if current user owns HDA.

model_class

alias of galaxy.model.HistoryDatasetAssociation

purge(hda, flush=True)[source]
tag_assoc

alias of galaxy.model.HistoryDatasetAssociationTagAssociation

text_data(hda, preview=True)[source]

Get data from text file, truncating if necessary.

class galaxy.managers.hdas.HDASerializer(app: galaxy.structured_app.StructuredApp)[source]

Bases: galaxy.managers.datasets.DatasetAssociationSerializer, galaxy.managers.taggable.TaggableSerializerMixin, galaxy.managers.annotatable.AnnotatableSerializerMixin

__init__(app: galaxy.structured_app.StructuredApp)[source]
add_serializers()[source]
model_manager_class

alias of HDAManager

serialize(hda, keys, user=None, **context)[source]

Override to hide information to users not able to access.

serialize_copied_from_ldda_id(item, key, **context)[source]

Serialize an id attribute of item.

serialize_display_apps(item, key, trans=None, **context)[source]

Return dictionary containing new-style display app urls.

serialize_old_display_applications(item, key, trans=None, **context)[source]

Return dictionary containing old-style display app urls.

serialize_urls(item, key, **context)[source]

Return web controller urls useful for this HDA.

Return a list of dictionaries with links to visualization pages for those visualizations that apply to this hda.

exception galaxy.managers.hdas.HistoryDatasetAssociationNoHistoryException[source]

Bases: Exception

galaxy.managers.hdcas module

Manager and Serializer for HDCAs.

HistoryDatasetCollectionAssociations (HDCAs) are datasets contained or created in a history.

class galaxy.managers.hdcas.DCASerializer(app: galaxy.structured_app.StructuredApp, dce_serializer=None)[source]

Bases: galaxy.managers.base.ModelSerializer

Base (abstract) Serializer class for HDCAs and LDCAs.

__init__(app: galaxy.structured_app.StructuredApp, dce_serializer=None)[source]
add_serializers()[source]
class galaxy.managers.hdcas.DCESerializer(app: galaxy.structured_app.StructuredApp)[source]

Bases: galaxy.managers.base.ModelSerializer

Serializer for DatasetCollectionElements.

__init__(app: galaxy.structured_app.StructuredApp)[source]
add_serializers()[source]
serialize_object(item, key, **context)[source]
class galaxy.managers.hdcas.DCSerializer(app: galaxy.structured_app.StructuredApp, dce_serializer=None)[source]

Bases: galaxy.managers.base.ModelSerializer

Serializer for DatasetCollections.

__init__(app: galaxy.structured_app.StructuredApp, dce_serializer=None)[source]
add_serializers()[source]
serialize_elements(item, key, **context)[source]
class galaxy.managers.hdcas.HDCAManager(app: galaxy.structured_app.MinimalManagerApp)[source]

Bases: galaxy.managers.base.ModelManager, galaxy.managers.secured.AccessibleManagerMixin, galaxy.managers.secured.OwnableManagerMixin, galaxy.managers.deletable.PurgableManagerMixin, galaxy.managers.taggable.TaggableManagerMixin, galaxy.managers.annotatable.AnnotatableManagerMixin

Interface/service object for interacting with HDCAs.

__init__(app: galaxy.structured_app.MinimalManagerApp)[source]

Set up and initialize other managers needed by hdas.

annotation_assoc

alias of galaxy.model.HistoryDatasetCollectionAssociationAnnotationAssociation

foreign_key_name = 'history_dataset_collection_association'
map_datasets(content, fn, *parents)[source]

Iterate over the datasets of a given collection, recursing into collections, and calling fn on each dataset.

Uses the same kwargs as contents above.

model_class

alias of galaxy.model.HistoryDatasetCollectionAssociation

tag_assoc

alias of galaxy.model.HistoryDatasetCollectionTagAssociation

update_attributes(content, payload: typing.Dict)[source]
class galaxy.managers.hdcas.HDCASerializer(app: galaxy.structured_app.StructuredApp)[source]

Bases: galaxy.managers.hdcas.DCASerializer, galaxy.managers.taggable.TaggableSerializerMixin, galaxy.managers.annotatable.AnnotatableSerializerMixin

Serializer for HistoryDatasetCollectionAssociations.

__init__(app: galaxy.structured_app.StructuredApp)[source]
add_serializers()[source]
generate_contents_url(item, key, **context)[source]
serialize_elements_datatypes(item, key, **context)[source]
serialize_job_state_summary(item, key, **context)[source]
galaxy.managers.hdcas.set_collection_attributes(dataset_element, *payload)[source]
galaxy.managers.hdcas.stream_dataset_collection(dataset_collection_instance, upstream_mod_zip=False, upstream_gzip=False)[source]

galaxy.managers.histories module

Manager and Serializer for histories.

Histories are containers for datasets or dataset collections created (or copied) by users over the course of an analysis.

class galaxy.managers.histories.HistoryDeserializer(app: galaxy.structured_app.MinimalManagerApp)[source]

Bases: galaxy.managers.sharable.SharableModelDeserializer, galaxy.managers.deletable.PurgableDeserializerMixin

Interface/service object for validating and deserializing dictionaries into histories.

__init__(app: galaxy.structured_app.MinimalManagerApp)[source]
add_deserializers()[source]
model_manager_class

alias of HistoryManager

class galaxy.managers.histories.HistoryExportView(app: galaxy.structured_app.MinimalManagerApp)[source]

Bases: object

__init__(app: galaxy.structured_app.MinimalManagerApp)[source]
get_exports(trans, history_id)[source]
get_ready_jeha(trans, history_id, jeha_id='latest')[source]
serialize(trans, history_id, jeha)[source]
class galaxy.managers.histories.HistoryFilters(app: galaxy.structured_app.MinimalManagerApp, **kwargs)[source]

Bases: galaxy.managers.sharable.SharableModelFilters, galaxy.managers.deletable.PurgableFiltersMixin

model_class

alias of galaxy.model.History

model_manager_class

alias of HistoryManager

class galaxy.managers.histories.HistoryManager(app: galaxy.structured_app.MinimalManagerApp, hda_manager: galaxy.managers.hdas.HDAManager, contents_manager: galaxy.managers.history_contents.HistoryContentsManager, contents_filters: galaxy.managers.history_contents.HistoryContentsFilters)[source]

Bases: galaxy.managers.sharable.SharableModelManager, galaxy.managers.deletable.PurgableManagerMixin, galaxy.managers.base.SortableManager

__init__(app: galaxy.structured_app.MinimalManagerApp, hda_manager: galaxy.managers.hdas.HDAManager, contents_manager: galaxy.managers.history_contents.HistoryContentsManager, contents_filters: galaxy.managers.history_contents.HistoryContentsFilters)[source]
annotation_assoc

alias of galaxy.model.HistoryAnnotationAssociation

by_user(user, current_history=None, **kwargs)[source]

Get all the histories for a given user (allowing anon users’ theirs) ordered by update time.

copy(history, user, **kwargs)[source]

Copy and return the given history.

foreign_key_name = 'history'
get_current(trans)[source]

Return the current history.

get_ready_history_export_file_path(trans, jeha) → str[source]

Serves the history export archive for use as a streaming response so the file doesn’t need to be loaded into memory.

get_sharing_extra_information(trans, item, users: typing.Set[galaxy.model.User], errors: typing.Set[str], option: typing.Optional[galaxy.schema.schema.SharingOptions] = None) → typing.Optional[galaxy.schema.schema.ShareWithExtra][source]

Returns optional extra information about the datasets of the history that can be accessed by the users.

is_history_shared_with(history, user) → bool[source]
is_owner(history, user, current_history=None, **kwargs)[source]

True if the current user is the owner of the given history.

legacy_serve_ready_history_export(trans, jeha)[source]
make_members_public(trans, item)[source]

Make the non-purged datasets in history public. Performs permissions check.

model_class

alias of galaxy.model.History

most_recent(user, filters=None, current_history=None, **kwargs)[source]

Return the most recently update history for the user.

If user is anonymous, return the current history. If the user is anonymous and the current history is deleted, return None.

non_ready_jobs(history)[source]

Return the currently running job objects associated with this history.

Where running is defined as new, waiting, queued, running, resubmitted, and upload.

parse_order_by(order_by_string, default=None)[source]

Return an ORM compatible order_by using the given string

purge(history, flush=True, **kwargs)[source]

Purge this history and all HDAs, Collections, and Datasets inside this history.

queue_history_export(trans, history, gzip=True, include_hidden=False, include_deleted=False, directory_uri=None, file_name=None)[source]
queue_history_import(trans, archive_type, archive_source)[source]
rating_assoc

alias of galaxy.model.HistoryRatingAssociation

set_current(trans, history)[source]

Set the current history.

set_current_by_id(trans, history_id)[source]

Set the current history by an id.

tag_assoc

alias of galaxy.model.HistoryTagAssociation

user_share_model

alias of galaxy.model.HistoryUserShareAssociation

class galaxy.managers.histories.HistorySerializer(app: galaxy.structured_app.MinimalManagerApp, hda_manager: galaxy.managers.hdas.HDAManager, hda_serializer: galaxy.managers.hdas.HDASerializer, history_contents_serializer: galaxy.managers.history_contents.HistoryContentsSerializer)[source]

Bases: galaxy.managers.sharable.SharableModelSerializer, galaxy.managers.deletable.PurgableSerializerMixin

Interface/service object for serializing histories into dictionaries.

SINGLE_CHAR_ABBR = 'h'
__init__(app: galaxy.structured_app.MinimalManagerApp, hda_manager: galaxy.managers.hdas.HDAManager, hda_serializer: galaxy.managers.hdas.HDASerializer, history_contents_serializer: galaxy.managers.history_contents.HistoryContentsSerializer)[source]
add_serializers()[source]
model_manager_class

alias of HistoryManager

serialize_contents(item, key, trans=None, user=None, **context)[source]
serialize_contents_active(item, key, **context)[source]

Return a dictionary keyed with ‘deleted’, ‘hidden’, and ‘active’ with values for each representing the count of contents in each state.

Note: counts for deleted and hidden overlap; In other words, a dataset that’s both deleted and hidden will be added to both totals.

serialize_contents_states(item, key, trans=None, **context)[source]

Return a dictionary containing the counts of all contents in each state keyed by the distinct states.

Note: does not include deleted/hidden contents.

serialize_history_state(item, key, **context)[source]

Returns the history state based on the states of the HDAs it contains.

serialize_state_counts(item, key, exclude_deleted=True, exclude_hidden=False, **context)[source]

Return a dictionary keyed to possible dataset states and valued with the number of datasets in this history that have those states.

serialize_state_ids(item, key, **context)[source]

Return a dictionary keyed to possible dataset states and valued with lists containing the ids of each HDA in that state.

galaxy.managers.history_contents module

Heterogenous lists/contents are difficult to query properly since unions are not easily made.

class galaxy.managers.history_contents.HistoryContentsFilters(app: galaxy.structured_app.MinimalManagerApp, **kwargs)[source]

Bases: galaxy.managers.base.ModelFilterParser, galaxy.managers.annotatable.AnnotatableFilterMixin, galaxy.managers.deletable.PurgableFiltersMixin, galaxy.managers.taggable.TaggableFilterMixin, galaxy.managers.tools.ToolFilterMixin

decode_type_id(type_id)[source]
model_class

alias of galaxy.model.HistoryDatasetAssociation

parse_type_id_list(type_id_list_string, sep=', ')[source]

Split type_id_list_string at sep.

class galaxy.managers.history_contents.HistoryContentsManager(app: galaxy.structured_app.MinimalManagerApp)[source]

Bases: galaxy.managers.base.SortableManager

__init__(app: galaxy.structured_app.MinimalManagerApp)[source]
active_counts(history)[source]

Return a dictionary keyed with ‘deleted’, ‘hidden’, and ‘active’ with values for each representing the count of contents in each state.

Note: counts for deleted and hidden overlap; In other words, a dataset that’s both deleted and hidden will be added to both totals.

common_columns = ('history_id', 'history_content_type', 'id', 'type_id', 'hid', 'extension', 'dataset_id', 'collection_id', 'name', 'state', 'deleted', 'purged', 'visible', 'create_time', 'update_time')
contained(container, filters=None, limit=None, offset=None, order_by=None, **kwargs)[source]

Returns non-subcontainer objects within container.

contained_class

alias of galaxy.model.HistoryDatasetAssociation

contained_class_manager_class

alias of galaxy.managers.hdas.HDAManager

contained_class_type_name = 'dataset'
contents(container, filters=None, limit=None, offset=None, order_by=None, **kwargs)[source]

Returns a list of both/all types of contents, filtered and in some order.

contents_count(container, filters=None, limit=None, offset=None, order_by=None, **kwargs)[source]

Returns a count of both/all types of contents, based on the given filters.

contents_query(container, filters=None, limit=None, offset=None, order_by=None, **kwargs)[source]

Returns the contents union query for subqueries, etc.

default_order_by = 'hid'
map_datasets(history, fn, **kwargs)[source]

Iterate over the datasets of a given history, recursing into collections, and calling fn on each dataset.

Uses the same kwargs as contents above.

parse_order_by(order_by_string, default=None)[source]

Return an ORM compatible order_by using the given string

static passes_filters(content, filters)[source]
root_container_class

alias of galaxy.model.History

state_counts(history)[source]

Return a dictionary containing the counts of all contents in each state keyed by the distinct states.

Note: does not include deleted/hidden contents.

subcontainer_class

alias of galaxy.model.HistoryDatasetCollectionAssociation

subcontainer_class_manager_class

alias of galaxy.managers.hdcas.HDCAManager

subcontainer_class_type_name = 'dataset_collection'
subcontainers(container, filters=None, limit=None, offset=None, order_by=None, **kwargs)[source]

Returns only the containers within container.

class galaxy.managers.history_contents.HistoryContentsSerializer(app: galaxy.structured_app.MinimalManagerApp, **kwargs)[source]

Bases: galaxy.managers.base.ModelSerializer, galaxy.managers.deletable.PurgableSerializerMixin

Interface/service object for serializing histories into dictionaries.

__init__(app: galaxy.structured_app.MinimalManagerApp, **kwargs)[source]
add_serializers()[source]
model_manager_class

alias of HistoryContentsManager

serialize_id_or_skip(item: typing.Any, key: str, **context)[source]

Serialize id or skip if attribute with key is not present.

galaxy.managers.interactivetool module

class galaxy.managers.interactivetool.InteractiveToolManager(app)[source]

Bases: object

Manager for dealing with InteractiveTools

__init__(app)[source]
access_entry_point_target(trans, entry_point_id)[source]
can_access_entry_point(trans, entry_point)[source]
can_access_entry_points(trans, entry_points)[source]
can_access_job(trans, job)[source]
configure_entry_point(job, tool_port=None, host=None, port=None, protocol=None)[source]
configure_entry_points(job, ports_dict)[source]
create_entry_points(job, tool, entry_points=None, flush=True)[source]
create_interactivetool(job, tool, entry_points)[source]
get_entry_point_path(trans, entry_point)[source]
get_entry_point_subdomain(trans, entry_point)[source]
get_nonterminal_for_user_by_trans(trans)[source]
remove_entry_point(entry_point, flush=True)[source]
remove_entry_points(entry_points)[source]
save_entry_point(entry_point)[source]

Writeout a key, key_type, token, value store that is used validating access

stop(trans, entry_point)[source]
target_if_active(trans, entry_point)[source]
class galaxy.managers.interactivetool.InteractiveToolSqlite(sqlite_filename, encode_id)[source]

Bases: object

__init__(sqlite_filename, encode_id)[source]
get(key, key_type)[source]
remove(**kwd)[source]

Remove entry from a key, key_type, token, value store that is can be used for coordinating with external resources. Remove entries that match all provided key=values

remove_entry_point(entry_point)[source]

Convenience method to easily remove an entry_point.

save(key, key_type, token, host, port, info=None)[source]

Writeout a key, key_type, token, value store that is can be used for coordinating with external resources.

save_entry_point(entry_point)[source]

Convenience method to easily save an entry_point.

galaxy.managers.jobs module

class galaxy.managers.jobs.JobLock[source]

Bases: pydantic.main.BaseModel

class galaxy.managers.jobs.JobManager(app: galaxy.structured_app.StructuredApp)[source]

Bases: object

__init__(app: galaxy.structured_app.StructuredApp)[source]
get_accessible_job(trans, decoded_job_id)[source]
job_lock() → galaxy.managers.jobs.JobLock[source]
stop(job, message=None)[source]
update_job_lock(job_lock: galaxy.managers.jobs.JobLock)[source]
class galaxy.managers.jobs.JobSearch(sa_session: galaxy.model.scoped_session.galaxy_scoped_session, hda_manager: galaxy.managers.hdas.HDAManager, dataset_collection_manager: galaxy.managers.collections.DatasetCollectionManager, ldda_manager: galaxy.managers.lddas.LDDAManager, id_encoding_helper: galaxy.security.idencoding.IdEncodingHelper)[source]

Bases: object

Search for jobs using tool inputs or other jobs

__init__(sa_session: galaxy.model.scoped_session.galaxy_scoped_session, hda_manager: galaxy.managers.hdas.HDAManager, dataset_collection_manager: galaxy.managers.collections.DatasetCollectionManager, ldda_manager: galaxy.managers.lddas.LDDAManager, id_encoding_helper: galaxy.security.idencoding.IdEncodingHelper)[source]
by_tool_input(trans, tool_id, tool_version, param=None, param_dump=None, job_state='ok')[source]

Search for jobs producing same results using the ‘inputs’ part of a tool POST.

galaxy.managers.jobs.fetch_job_states(sa_session, job_source_ids, job_source_types)[source]
galaxy.managers.jobs.get_output_name(tool, output, params)[source]
galaxy.managers.jobs.get_path_key(path_tuple)[source]
galaxy.managers.jobs.invocation_job_source_iter(sa_session, invocation_id)[source]
galaxy.managers.jobs.summarize_destination_params(trans, job)[source]

Produce a dict-ified version of job destination parameters ready for tabular rendering.

Precondition: the caller has verified the job is accessible to the user represented by the trans parameter.

galaxy.managers.jobs.summarize_invocation_jobs(invocation_id, job_summaries, implicit_collection_job_summaries, invocation_state, invocation_step_states)[source]
galaxy.managers.jobs.summarize_job_metrics(trans, job)[source]

Produce a dict-ified version of job metrics ready for tabular rendering.

Precondition: the caller has verified the job is accessible to the user represented by the trans parameter.

galaxy.managers.jobs.summarize_job_outputs(job: galaxy.model.Job, tool, params, security)[source]
galaxy.managers.jobs.summarize_job_parameters(trans, job)[source]

Produce a dict-ified version of job parameters ready for tabular rendering.

Precondition: the caller has verified the job is accessible to the user represented by the trans parameter.

galaxy.managers.jobs.summarize_jobs_to_dict(sa_session, jobs_source)[source]

Produce a summary of jobs for job summary endpoints.

Parameters:jobs_source (a Job or ImplicitCollectionJobs or None) – the object to summarize
Return type:dict
Returns:dictionary containing job summary information
galaxy.managers.jobs.view_show_job(trans, job, full: bool) → typing.Dict[source]

galaxy.managers.lddas module

class galaxy.managers.lddas.LDDAManager(app: galaxy.structured_app.MinimalManagerApp)[source]

Bases: galaxy.managers.datasets.DatasetAssociationManager

A fairly sparse manager for LDDAs.

model_class

alias of galaxy.model.LibraryDatasetDatasetAssociation

__init__(app: galaxy.structured_app.MinimalManagerApp)[source]

Set up and initialize other managers needed by lddas.

get(trans, id, check_accessible=True)[source]

galaxy.managers.libraries module

Manager and Serializer for libraries.

class galaxy.managers.libraries.LibraryManager[source]

Bases: object

Interface/service object for interacting with libraries.

check_accessible(trans, library)[source]

Check whether the library is accessible to current user.

create(trans, name, description='', synopsis='')[source]

Create a new library.

delete(trans, library, undelete=False)[source]

Mark given library deleted/undeleted based on the flag.

get(trans, decoded_library_id, check_accessible=True)[source]

Get the library from the DB.

Parameters:
  • decoded_library_id (int) – decoded library id
  • check_accessible (bool) – flag whether to check that user can access item
Returns:

the requested library

Return type:

galaxy.model.Library

get_access_roles(trans, library)[source]

Load access roles for all library permissions

get_add_roles(trans, library)[source]

Load add roles for all library permissions

get_current_roles(trans, library)[source]

Load all permissions currently related to the given library.

Parameters:library (galaxy.model.Library) – the model object
Return type:dictionary
Returns:dict of current roles for all available permission types
get_library_dict(trans, library, prefetched_ids=None)[source]

Return library data in the form of a dictionary.

Parameters:
  • library (galaxy.model.Library) – library
  • prefetched_ids (dict) – dict of 3 sets with available actions for user’s accessible libraries and a set of ids of all public libraries. These are used for limiting the number of queries when dictifying a set of libraries.
Returns:

dict with data about the library

Return type:

dictionary

get_manage_roles(trans, library)[source]

Load manage roles for all library permissions

get_modify_roles(trans, library)[source]

Load modify roles for all library permissions

is_public(trans, library)[source]

Return true if lib is public.

list(trans, deleted: typing.Optional[bool] = False)[source]

Return a list of libraries from the DB.

Parameters:deleted (boolean (optional)) – if True, show only deleted libraries, if False show only non-deleted
Returns:query that will emit all accessible libraries
Return type:sqlalchemy query
Returns:dict of 3 sets with available actions for user’s accessible libraries and a set of ids of all public libraries. These are used for limiting the number of queries when dictifying the libraries later on.
Return type:dict
make_public(trans, library)[source]

Makes the given library public (removes all access roles)

secure(trans, library, check_accessible=True)[source]

Check if library is accessible to user.

Parameters:
  • library (galaxy.model.Library) – library
  • check_accessible (bool) – flag whether to check that user can access library
Returns:

the original library

Return type:

galaxy.model.Library

set_permission_roles(trans, library, access_roles, modify_roles, manage_roles, add_roles)[source]

Set permissions on the given library.

update(trans, library, name=None, description=None, synopsis=None)[source]

Update the given library

galaxy.managers.libraries.get_containing_library_from_library_dataset(trans, library_dataset)[source]

Given a library_dataset, get the containing library

galaxy.managers.library_datasets module

Manager and Serializer for library datasets.

class galaxy.managers.library_datasets.LibraryDatasetsManager(app: galaxy.structured_app.MinimalManagerApp)[source]

Bases: galaxy.managers.datasets.DatasetAssociationManager

Interface/service object for interacting with library datasets.

__init__(app: galaxy.structured_app.MinimalManagerApp)[source]
check_accessible(trans, ld)[source]

Check whether the current user has permissions to access library dataset.

Parameters:ld (galaxy.model.LibraryDataset) – library dataset
Returns:the original library dataset
Return type:galaxy.model.LibraryDataset
Raises:ObjectNotFound
check_modifiable(trans, ld)[source]

Check whether the current user has permissions to modify library dataset.

Parameters:ld (galaxy.model.LibraryDataset) – library dataset
Returns:the original library dataset
Return type:galaxy.model.LibraryDataset
Raises:ObjectNotFound
get(trans, decoded_library_dataset_id, check_accessible=True)[source]

Get the library dataset from the DB.

Parameters:
  • decoded_library_dataset_id (int) – decoded library dataset id
  • check_accessible (bool) – flag whether to check that user can access item
Returns:

the requested library dataset

Return type:

galaxy.model.LibraryDataset

model_class

alias of galaxy.model.LibraryDatasetDatasetAssociation

secure(trans, ld, check_accessible=True, check_ownership=False)[source]

Check if library dataset is accessible to current user or the user is an admin.

Parameters:
Returns:

the original library dataset

Return type:

galaxy.model.LibraryDataset

serialize(trans, ld)[source]

Serialize the library dataset into a dictionary.

update(trans, ld, payload)[source]

Update the given library dataset - the latest linked ldda. Updating older lddas (versions) is not allowed.

Parameters:
  • ld (LibraryDataset) – library dataset to change
  • payload (dict) – dictionary structure containing:: :param name: new ld’s name, must be longer than 0 :type name: str :param misc_info: new ld’s misc info :type misc_info: str :param file_ext: new ld’s extension, must exist in the Galaxy registry :type file_ext: str :param genome_build: new ld’s genome build :type genome_build: str :param tags: list of dataset tags :type tags: list
Returns:

the changed library dataset

Return type:

galaxy.model.LibraryDataset

galaxy.managers.licenses module

class galaxy.managers.licenses.LicenseMetadataModel[source]

Bases: pydantic.main.BaseModel

class galaxy.managers.licenses.LicensesManager[source]

Bases: object

__init__()[source]
get(uri)[source]
get_license_by_id(id: str) → galaxy.managers.licenses.LicenseMetadataModel[source]
get_licenses() → typing.List[galaxy.managers.licenses.LicenseMetadataModel][source]
index()[source]

galaxy.managers.markdown_parse module

Utilities for parsing “Galaxy Flavored Markdown”.

See markdown_util.py for loading objects and interacting with the rest of Galaxy. This file is meant to be relatively self contained and just used to “parse” and validate Galaxy Markdown. Keeping things isolated to allow re-use of these utilities in other projects (e.g. gxformat2).

galaxy.managers.markdown_parse.validate_galaxy_markdown(galaxy_markdown, internal=True)[source]

Validate the supplied markdown and throw an ValueError with reason if invalid.

galaxy.managers.markdown_util module

Utilities defining “Galaxy Flavored Markdown”.

This is an extension of markdown designed to allow rendering Galaxy object references.

The core “Galaxy Flavored Markdown” format should just reference objects by encoded IDs - but preprocessing should allow for instance workflow objects to be referenced relative to the workflow (inputs, outputs, steps, etc..) and potential history flavor would allow objects to be referenced by HID. This second idea is unimplemented, it is just an example of the general concept of context specific processing.

galaxy.managers.markdown_util.internal_galaxy_markdown_to_pdf(trans, internal_galaxy_markdown: str, document_type: galaxy.schema.PdfDocumentType) → bytes[source]
galaxy.managers.markdown_util.ready_galaxy_markdown_for_export(trans, internal_galaxy_markdown)[source]

Fill in details needed to render Galaxy flavored markdown.

Take it from a minimal internal version to an externally render-able version with more details populated and actual IDs replaced with encoded IDs to render external links. Return expanded markdown and extra data useful for rendering custom container tags.

galaxy.managers.markdown_util.ready_galaxy_markdown_for_import(trans, external_galaxy_markdown)[source]

Convert from encoded IDs to decoded numeric IDs for storing in the DB.

galaxy.managers.markdown_util.resolve_invocation_markdown(trans, invocation, workflow_markdown)[source]

Resolve invocation objects to convert markdown to ‘internal’ representation.

Replace references to abstract workflow parts with actual galaxy object IDs corresponding to the actual executed workflow. For instance:

convert output=name -to- history_dataset_id=<id> | history_dataset_collection_id=<id> convert input=name -to- history_dataset_id=<id> | history_dataset_collection_id=<id> convert step=name -to- job_id=<id>

Also expand/convert workflow invocation specific container sections into actual Galaxy markdown - these containers include: invocation_inputs, invocation_outputs, invocation_workflow. Hopefully this list will be expanded to include invocation_qc.

galaxy.managers.pages module

Manager and Serializers for Pages.

Pages are markup created and saved by users that can contain Galaxy objects (such as datasets) and are often used to describe or present an analysis from within Galaxy.

class galaxy.managers.pages.PageContentProcessor(trans, render_embed_html_fn: typing.Callable)[source]

Bases: html.parser.HTMLParser

Processes page content to produce HTML that is suitable for display. For now, processor renders embedded objects.

__init__(trans, render_embed_html_fn: typing.Callable)[source]
bare_ampersand = re.compile('&(?!#\\d+;|#x[0-9a-fA-F]+;|\\w+;)')
elements_no_end_tag = {'track', 'base', 'hr', 'source', 'wbr', 'basefont', 'meta', 'img', 'link', 'param', 'area', 'br', 'isindex', 'input', 'keygen', 'command', 'embed', 'frame', 'col'}
feed(data)[source]
handle_charref(ref)[source]
handle_comment(text)[source]
handle_data(text)[source]

Called for each block of plain text

Called outside of any tag and not containing any character or entity references. Store the original text verbatim.

handle_decl(text)[source]
handle_endtag(tag)[source]

Called for each end tag

E.g. for </pre>, tag will be ‘pre’

handle_entityref(ref)[source]
handle_pi(text)[source]
handle_starttag(tag, attrs)[source]

Called for each start tag

attrs is a list of (attr, value) tuples, e.g. for <pre class=’screen’>, tag=’pre’, attrs=[(‘class’, ‘screen’)]

output()[source]

Return processed HTML as a single string

reset()[source]
class galaxy.managers.pages.PageDeserializer(app: galaxy.structured_app.MinimalManagerApp)[source]

Bases: galaxy.managers.sharable.SharableModelDeserializer

Interface/service object for validating and deserializing dictionaries into pages.

__init__(app: galaxy.structured_app.MinimalManagerApp)[source]
add_deserializers()[source]
model_manager_class

alias of PageManager

class galaxy.managers.pages.PageManager(app: galaxy.structured_app.MinimalManagerApp)[source]

Bases: galaxy.managers.sharable.SharableModelManager, galaxy.model.item_attrs.UsesAnnotations

Provides operations for managing a Page.

__init__(app: galaxy.structured_app.MinimalManagerApp)[source]
annotation_assoc

alias of galaxy.model.PageAnnotationAssociation

create(trans, payload)[source]
foreign_key_name = 'page'
model_class

alias of galaxy.model.Page

rating_assoc

alias of galaxy.model.PageRatingAssociation

rewrite_content_for_export(trans, as_dict)[source]
rewrite_content_for_import(trans, content, content_format: str)[source]
save_new_revision(trans, page, payload)[source]
tag_assoc

alias of galaxy.model.PageTagAssociation

user_share_model

alias of galaxy.model.PageUserShareAssociation

class galaxy.managers.pages.PageSerializer(app: galaxy.structured_app.MinimalManagerApp)[source]

Bases: galaxy.managers.sharable.SharableModelSerializer

Interface/service object for serializing pages into dictionaries.

SINGLE_CHAR_ABBR = 'p'
__init__(app: galaxy.structured_app.MinimalManagerApp)[source]
add_serializers()[source]
model_manager_class

alias of PageManager

galaxy.managers.pages.get_page_identifiers(item_id, app)[source]
galaxy.managers.pages.placeholderRenderForEdit(trans: galaxy.managers.context.ProvidesHistoryContext, item_class, item_id)[source]
galaxy.managers.pages.placeholderRenderForSave(trans: galaxy.managers.context.ProvidesHistoryContext, item_class, item_id, encode=False)[source]

galaxy.managers.ratable module

Mixins for Ratable model managers and serializers.

class galaxy.managers.ratable.RatableDeserializerMixin[source]

Bases: object

add_deserializers()[source]
deserialize_rating(item, key, val, user=None, **context)[source]
class galaxy.managers.ratable.RatableFilterMixin[source]

Bases: object

class galaxy.managers.ratable.RatableManagerMixin[source]

Bases: object

rate(item, user, value, flush=True)[source]

Updates or creates a rating for this item and user. Returns the rating

rating(item, user, as_int=True)[source]

Returns the integer rating given to this item by the user.

Returns the full rating model if as_int is False.

ratings(item)[source]

Returns a list of all rating values given to this item.

ratings_avg(item)[source]

Returns the average of all ratings given to this item.

ratings_count(item)[source]

Returns the number of ratings given to this item.

class galaxy.managers.ratable.RatableSerializerMixin[source]

Bases: object

add_serializers()[source]
serialize_community_rating(item, key, **context)[source]
Returns a dictionary containing:
average the (float) average of all ratings of this object count the number of ratings
serialize_user_rating(item, key, user=None, **context)[source]

Returns the integer rating given to this item by the user.

galaxy.managers.rbac_secured module

class galaxy.managers.rbac_secured.AccessDatasetRBACPermission(app)[source]

Bases: galaxy.managers.rbac_secured.DatasetRBACPermission

A class to manage access permissions on a dataset.

An user must have all the Roles of all the access permissions associated with a dataset in order to access it.

action_name = 'access'
grant(item, user)[source]
is_permitted(dataset, user, trans=None)[source]
is_public(dataset)[source]
permission_failed_error_class

alias of DatasetAccessPermissionFailedException

revoke(item, user)[source]
set_private(dataset, user, flush=True)[source]
exception galaxy.managers.rbac_secured.DatasetAccessPermissionFailedException(err_msg=None, type='info', **extra_error_info)[source]

Bases: galaxy.managers.rbac_secured.RBACPermissionFailedException

exception galaxy.managers.rbac_secured.DatasetManagePermissionFailedException(err_msg=None, type='info', **extra_error_info)[source]

Bases: galaxy.managers.rbac_secured.RBACPermissionFailedException

class galaxy.managers.rbac_secured.DatasetRBACPermission(app)[source]

Bases: galaxy.managers.rbac_secured.RBACPermission

Base class for the manage and access RBAC permissions used by dataset security.

The DatasetPermissions used by the RBAC agent are associations between a Dataset and a single Role.

DatasetPermissions are typed (but not polymorphic themselves) by a string ‘action’. There are two types:

  • manage permissions : can a role manage the permissions on a dataset
  • access : can a role read/look at/copy a dataset
action_name = None
by_dataset(dataset)[source]
by_role(dataset, role)[source]
by_roles(dataset, roles)[source]
clear(dataset, flush=True)[source]
permissions_class

alias of galaxy.model.DatasetPermissions

set(dataset, roles, flush=True)[source]
class galaxy.managers.rbac_secured.ManageDatasetRBACPermission(app)[source]

Bases: galaxy.managers.rbac_secured.DatasetRBACPermission

A class that controls the dataset permissions that control who can manage that dataset’s permissions.

When checking permissions for a user, if any of the user’s roles have permission on the dataset

action_name = 'manage permissions'
grant(dataset, user, flush=True)[source]
is_permitted(dataset, user, trans=None)[source]
permission_failed_error_class

alias of DatasetManagePermissionFailedException

revoke(dataset, user, flush=True)[source]
class galaxy.managers.rbac_secured.RBACPermission(app)[source]

Bases: object

Base class for wrangling/controlling the permissions ORM models (Permissions, Roles) that control which users can perform certain actions on their associated models (Libraries, Datasets).

__init__(app)[source]
error_unless_permitted(item, user, trans=None)[source]
grant(item, user, flush=True)[source]
is_permitted(item, user, trans=None)[source]
permission_failed_error_class

alias of RBACPermissionFailedException

revoke(item, user, flush=True)[source]
session()[source]
exception galaxy.managers.rbac_secured.RBACPermissionFailedException(err_msg=None, type='info', **extra_error_info)[source]

Bases: galaxy.exceptions.InsufficientPermissionsException

galaxy.managers.rbac_secured.iterable_has_all(iterable, has_these)[source]

galaxy.managers.roles module

Manager and Serializer for Roles.

class galaxy.managers.roles.RoleManager(app: galaxy.structured_app.BasicSharedApp)[source]

Bases: galaxy.managers.base.ModelManager

Business logic for roles.

create_role(trans: galaxy.managers.context.ProvidesUserContext, role_definition_model) → galaxy.model.Role[source]
foreign_key_name = 'role'
get(trans: galaxy.managers.context.ProvidesUserContext, decoded_role_id)[source]

Method loads the role from the DB based on the given role id.

Parameters:decoded_role_id (int) – id of the role to load from the DB
Returns:the loaded Role object
Return type:galaxy.model.Role
Raises:InconsistentDatabase, RequestParameterInvalidException, InternalServerError
group_assoc

alias of galaxy.model.GroupRoleAssociation

list_displayable_roles(trans: galaxy.managers.context.ProvidesUserContext)[source]
model_class

alias of galaxy.model.Role

user_assoc

alias of galaxy.model.UserRoleAssociation

galaxy.managers.secured module

Accessible models can be read and copied but not modified or deleted.

Owned models can be modified and deleted.

class galaxy.managers.secured.AccessibleManagerMixin[source]

Bases: object

A security interface to check if a User can read/view an item’s.

This can also be thought of as ‘read but not modify’ privileges.

by_id(id: int)[source]
error_unless_accessible(item, user, **kwargs)[source]

Raise an error if the item is NOT accessible to user, otherwise return the item.

Raises:exceptions.ItemAccessibilityException
filter_accessible(user, **kwargs)[source]

Return a list of items accessible to the user.

get_accessible(id, user, **kwargs)[source]

Return the item with the given id if it’s accessible to user, otherwise raise an error.

Raises:exceptions.ItemAccessibilityException
is_accessible(item, user, **kwargs)[source]

Return True if the item accessible to user.

list_accessible(user, **kwargs)[source]

Return a list of items accessible to the user, raising an error if ANY are inaccessible.

Raises:exceptions.ItemAccessibilityException
class galaxy.managers.secured.OwnableManagerMixin[source]

Bases: object

A security interface to check if a User is an item’s owner.

Some resources are associated with the User that created or imported them and these Users can be considered the models’ owner.

This can also be thought of as write/edit privileges.

by_id(id: int)[source]
error_unless_owner(item, user, **kwargs)[source]

Raise an error if the item is NOT owned by user, otherwise return the item.

Raises:exceptions.ItemAccessibilityException
filter_owned(user, **kwargs)[source]

Return a list of items owned by the user.

get_owned(id, user, **kwargs)[source]

Return the item with the given id if owned by the user, otherwise raise an error.

Raises:exceptions.ItemOwnershipException
is_owner(item, user, **kwargs)[source]

Return True if user owns the item.

list_owned(user, **kwargs)[source]

Return a list of items owned by the user, raising an error if ANY are not.

Raises:exceptions.ItemAccessibilityException

galaxy.managers.session module

class galaxy.managers.session.GalaxySessionManager(model: galaxy.model.base.SharedModelMapping)[source]

Bases: object

Manages GalaxySession.

__init__(model: galaxy.model.base.SharedModelMapping)[source]
get_session_from_session_key(session_key: str)[source]

Returns GalaxySession if session_key is valid.

galaxy.managers.sharable module

Superclass Manager and Serializers for Sharable objects.

A sharable Galaxy object:
has an owner/creator User is sharable with other, specific Users is importable (copyable) by users that have access has a slug which can be used as a link to view the resource can be published effectively making it available to all other Users can be rated
class galaxy.managers.sharable.SharableModelDeserializer(app: galaxy.structured_app.MinimalManagerApp, **kwargs)[source]

Bases: galaxy.managers.base.ModelDeserializer, galaxy.managers.taggable.TaggableDeserializerMixin, galaxy.managers.annotatable.AnnotatableDeserializerMixin, galaxy.managers.ratable.RatableDeserializerMixin

__init__(app: galaxy.structured_app.MinimalManagerApp, **kwargs)[source]
add_deserializers()[source]
deserialize_importable(item, key, val, **context)[source]
deserialize_published(item, key, val, **context)[source]
deserialize_users_shared_with(item, key, val, **context)[source]

Accept a list of encoded user_ids, validate them as users, and then add or remove user shares in order to update the users_shared_with to match the given list finally returning the new list of shares.

class galaxy.managers.sharable.SharableModelFilters(app: galaxy.structured_app.MinimalManagerApp, **kwargs)[source]

Bases: galaxy.managers.base.ModelFilterParser, galaxy.managers.taggable.TaggableFilterMixin, galaxy.managers.annotatable.AnnotatableFilterMixin, galaxy.managers.ratable.RatableFilterMixin

class galaxy.managers.sharable.SharableModelManager(app: galaxy.structured_app.MinimalManagerApp)[source]

Bases: galaxy.managers.base.ModelManager, galaxy.managers.secured.OwnableManagerMixin, galaxy.managers.secured.AccessibleManagerMixin, galaxy.managers.taggable.TaggableManagerMixin, galaxy.managers.annotatable.AnnotatableManagerMixin, galaxy.managers.ratable.RatableManagerMixin

SINGLE_CHAR_ABBR = None
__init__(app: galaxy.structured_app.MinimalManagerApp)[source]
by_user(user, filters=None, **kwargs)[source]

Return list for all items (of model_class type) associated with the given user.

create_unique_slug(item, flush=True)[source]

Set a new, unique slug on the item.

get_share_assocs(item, user=None)[source]

Get the UserShareAssociations for the item.

Optionally send in user to test for a single match.

get_sharing_extra_information(trans, item, users: typing.Set[galaxy.model.User], errors: typing.Set[str], option: typing.Optional[galaxy.schema.schema.SharingOptions] = None) → typing.Optional[galaxy.schema.schema.ShareWithExtra][source]

Returns optional extra information about the shareability of the given item.

This function should be overridden in the particular manager class that wants to provide the extra information, otherwise, it will be None by default.

get_unique_slug(item)[source]

Returns a slug that is unique among user’s importable items for item’s class.

is_accessible(item, user, **kwargs)[source]

If the item is importable, is owned by user, or (the valid) user is in ‘users shared with’ list for the item: return True.

is_owner(item, user, **kwargs)[source]

Return true if this sharable belongs to user (or user is an admin).

is_valid_slug(slug)[source]

Returns true if slug is valid.

list_published(filters=None, **kwargs)[source]

Return a list of all published items.

list_shared_with(user, filters=None, order_by=None, limit=None, offset=None, **kwargs)[source]

Return a list of those models shared with a particular user.

make_importable(item, flush=True)[source]

Makes item accessible–viewable and importable–and sets item’s slug. Does not flush/commit changes, however. Item must have name, user, importable, and slug attributes.

make_members_public(trans, item)[source]

Make potential elements of this item public.

This method must be overridden in managers that need to change permissions of internal elements contained associated with the given item.

make_non_importable(item, flush=True)[source]

Makes item accessible–viewable and importable–and sets item’s slug. Does not flush/commit changes, however. Item must have name, user, importable, and slug attributes.

publish(item, flush=True)[source]

Set both the importable and published flags on item to True.

set_slug(item, new_slug, user, flush=True)[source]

Validate and set the new slug for item.

share_with(item, user: galaxy.model.User, flush: bool = True)[source]

Get or create a share for the given user.

unpublish(item, flush=True)[source]

Set the published flag on item to False.

unshare_with(item, user: galaxy.model.User, flush: bool = True)[source]

Delete a user share from the database.

update_current_sharing_with_users(item, new_users_shared_with: typing.Set[galaxy.model.User], flush=True)[source]

Updates the currently list of users this item is shared with by adding new users and removing missing ones.

class galaxy.managers.sharable.SharableModelSerializer(app, **kwargs)[source]

Bases: galaxy.managers.base.ModelSerializer, galaxy.managers.taggable.TaggableSerializerMixin, galaxy.managers.annotatable.AnnotatableSerializerMixin, galaxy.managers.ratable.RatableSerializerMixin

SINGLE_CHAR_ABBR = None
__init__(app, **kwargs)[source]
add_serializers()[source]
serialize_title(item, key, **context)[source]
serialize_username_and_slug(item, key, **context)[source]
serialize_users_shared_with(item, key, user=None, **context)[source]

Returns a list of encoded ids for users the item has been shared.

Skipped if the requesting user is not the owner.

class galaxy.managers.sharable.SlugBuilder[source]

Bases: object

Builder for creating slugs out of items.

create_item_slug(sa_session, item) → bool[source]

Create/set item slug.

Slug is unique among user’s importable items for item’s class.

Parameters:
  • sa_session – Database session context.
  • item ([type]) – The item to create/update its slug.
Returns:

Returns true if item’s slug was set/changed; false otherwise.

Return type:

bool

galaxy.managers.taggable module

Mixins for Taggable model managers and serializers.

class galaxy.managers.taggable.TaggableDeserializerMixin[source]

Bases: object

add_deserializers()[source]
deserialize_tags(item, key, val, user=None, **context)[source]

Make sure val is a valid list of tag strings and assign them.

Note: this will erase any previous tags.

class galaxy.managers.taggable.TaggableFilterMixin[source]

Bases: object

create_tag_filter(attr, op, val)[source]
valid_ops = ('eq', 'contains', 'has')
class galaxy.managers.taggable.TaggableManagerMixin[source]

Bases: object

get_tags(item)[source]

Return a list of tag strings.

set_tags(item, new_tags, user=None)[source]

Set an item’s tags from a list of strings.

class galaxy.managers.taggable.TaggableSerializerMixin[source]

Bases: object

add_serializers()[source]
serialize_tags(item, key, **context)[source]

Return tags as a list of strings.

galaxy.managers.tools module

class galaxy.managers.tools.DynamicToolManager(app: galaxy.structured_app.BasicSharedApp)[source]

Bases: galaxy.managers.base.ModelManager

Manages dynamic tools stored in Galaxy’s database.

create_tool(trans, tool_payload, allow_load=True)[source]
deactivate(dynamic_tool)[source]
get_tool_by_id(object_id)[source]
get_tool_by_tool_id(tool_id)[source]
get_tool_by_uuid(uuid: typing.Union[uuid.UUID, str, NoneType])[source]
list_tools(active=True)[source]
model_class

alias of galaxy.model.DynamicTool

class galaxy.managers.tools.ToolFilterMixin[source]

Bases: object

create_tool_filter(attr, op, val)[source]

galaxy.managers.users module

Manager and Serializer for Users.

class galaxy.managers.users.AdminUserFilterParser(app: galaxy.structured_app.MinimalManagerApp, **kwargs)[source]

Bases: galaxy.managers.base.ModelFilterParser, galaxy.managers.deletable.PurgableFiltersMixin

model_class

alias of galaxy.model.User

model_manager_class

alias of UserManager

class galaxy.managers.users.CurrentUserSerializer(app: galaxy.structured_app.MinimalManagerApp)[source]

Bases: galaxy.managers.users.UserSerializer

model_manager_class

alias of UserManager

serialize(user, keys, **kwargs)[source]

Override to return at least some usage info if user is anonymous.

serialize_current_anonymous_user(user, keys, trans=None, **kwargs)[source]
class galaxy.managers.users.UserDeserializer(app: galaxy.structured_app.MinimalManagerApp, **kwargs)[source]

Bases: galaxy.managers.base.ModelDeserializer

Service object for validating and deserializing dictionaries that update/alter users.

add_deserializers()[source]
deserialize_username(item, key, username, trans=None, **context)[source]
model_manager_class

alias of UserManager

class galaxy.managers.users.UserManager(app: galaxy.structured_app.BasicSharedApp)[source]

Bases: galaxy.managers.base.ModelManager, galaxy.managers.deletable.PurgableManagerMixin

__init__(app: galaxy.structured_app.BasicSharedApp)[source]
activate(user)[source]
admins(filters=None, **kwargs)[source]

Return a list of admin Users.

by_api_key(api_key, sa_session=None)[source]

Find a user by API key.

by_email(email, filters=None, **kwargs)[source]

Find a user by their email.

by_id(user_id)[source]
change_password(trans, password=None, confirm=None, token=None, id=None, current=None)[source]

Allows to change a user password with a token.

check_master_api_key(api_key)[source]
create(email=None, username=None, password=None, **kwargs)[source]

Create a new user.

create_api_key(user)[source]

Create and return an API key for user.

current_user(trans)[source]
default_permissions(user)[source]
delete(user, flush=True)[source]

Mark the given user deleted.

error_if_anonymous(user, msg='Log-in required', **kwargs)[source]

Raise an error if user is anonymous.

error_unless_admin(user, msg='Administrators only', **kwargs)[source]

Raise an error if user is not an admin.

Raises:exceptions.AdminRequiredException – if user is not an admin.
foreign_key_name = 'user'
get_or_create_valid_api_key(user)[source]

Return this most recent APIKey for this user or create one if none have been created.

get_reset_token(trans, email)[source]
get_user_by_identity(identity)[source]

Get user by username or email.

is_admin(user, trans=None)[source]

Return True if this user is an admin (or session is authenticated as admin).

Do not pass trans to simply check if an existing user object is an admin user, pass trans when checking permissions.

is_anonymous(user)[source]

Return True if user is anonymous.

preferences(user)[source]
private_role(user)[source]
purge(user, flush=True)[source]

Purge the given user. They must have the deleted flag already.

quota(user, total=False)[source]
register(trans, email=None, username=None, password=None, confirm=None, subscribe=False)[source]

Register a new user.

send_activation_email(trans, email, username)[source]

Send the verification email containing the activation link to the user’s email.

send_reset_email(trans, payload, **kwd)[source]

Reset the user’s password. Send an email with token that allows a password change.

send_subscription_email(email)[source]
sharing_roles(user)[source]
tags_used(user, tag_models=None)[source]

Return a list of distinct ‘user_tname:user_value’ strings that the given user has used.

undelete(user, flush=True)[source]

Remove the deleted flag for the given user.

user_can_do_run_as(user) → bool[source]
valid_api_key(user)[source]

Return this most recent APIKey for this user or None if none have been created.

class galaxy.managers.users.UserSerializer(app: galaxy.structured_app.MinimalManagerApp)[source]

Bases: galaxy.managers.base.ModelSerializer, galaxy.managers.deletable.PurgableSerializerMixin

__init__(app: galaxy.structured_app.MinimalManagerApp)[source]

Convert a User and associated data to a dictionary representation.

add_serializers()[source]
model_manager_class

alias of UserManager

galaxy.managers.visualizations module

Manager and Serializers for Visualizations.

Visualizations are saved configurations/variables used to reproduce a specific view in a Galaxy visualization.

class galaxy.managers.visualizations.VisualizationManager(app: galaxy.structured_app.MinimalManagerApp)[source]

Bases: galaxy.managers.sharable.SharableModelManager

Handle operations outside and between visualizations and other models.

model_class

alias of galaxy.model.Visualization

foreign_key_name = 'visualization'
user_share_model

alias of galaxy.model.VisualizationUserShareAssociation

tag_assoc

alias of galaxy.model.VisualizationTagAssociation

annotation_assoc

alias of galaxy.model.VisualizationAnnotationAssociation

rating_assoc

alias of galaxy.model.VisualizationRatingAssociation

class galaxy.managers.visualizations.VisualizationSerializer(app: galaxy.structured_app.MinimalManagerApp)[source]

Bases: galaxy.managers.sharable.SharableModelSerializer

Interface/service object for serializing visualizations into dictionaries.

model_manager_class

alias of VisualizationManager

SINGLE_CHAR_ABBR = 'v'
__init__(app: galaxy.structured_app.MinimalManagerApp)[source]
add_serializers()[source]
class galaxy.managers.visualizations.VisualizationDeserializer(app)[source]

Bases: galaxy.managers.sharable.SharableModelDeserializer

Interface/service object for validating and deserializing dictionaries into visualizations.

model_manager_class

alias of VisualizationManager

__init__(app)[source]
add_deserializers()[source]

galaxy.managers.workflows module

class galaxy.managers.workflows.CreatedWorkflow(stored_workflow, workflow, missing_tools)

Bases: tuple

missing_tools

Alias for field number 2

stored_workflow

Alias for field number 0

workflow

Alias for field number 1

class galaxy.managers.workflows.Format2ConverterGalaxyInterface[source]

Bases: gxformat2.interface.ImporterGalaxyInterface

import_workflow(workflow, **kwds)[source]
exception galaxy.managers.workflows.MissingToolsException(workflow, errors)[source]

Bases: galaxy.exceptions.MessageException

__init__(workflow, errors)[source]
class galaxy.managers.workflows.RawWorkflowDescription(as_dict, workflow_path=None)[source]

Bases: object

__init__(as_dict, workflow_path=None)[source]
class galaxy.managers.workflows.RefactorRequest[source]

Bases: galaxy.workflow.refactor.schema.RefactorActions

class galaxy.managers.workflows.RefactorResponse[source]

Bases: pydantic.main.BaseModel

class Config[source]

Bases: object

json_dumps(*args, **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, and we ‘fix’ it before re-encoding.

class galaxy.managers.workflows.WorkflowContentsManager(app: galaxy.structured_app.MinimalManagerApp)[source]

Bases: galaxy.model.item_attrs.UsesAnnotations

__init__(app: galaxy.structured_app.MinimalManagerApp)[source]
build_workflow_from_raw_description(trans, raw_workflow_description, workflow_create_options, source=None, add_to_menu=False, hidden=False)[source]
do_refactor(trans, stored_workflow, refactor_request)[source]

Apply supplied actions to stored_workflow.latest_workflow to build a new version.

ensure_raw_description(dict_or_raw_description)[source]
get_all_tools(workflow)[source]
static get_step_map_over(current_step, steps)[source]

Given a tool step and its input steps guess that maximum level of mapping over. All data outputs of a step need to be mapped over to this level.

normalize_workflow_format(trans, as_dict)[source]

Process incoming workflow descriptions for consumption by other methods.

Currently this mostly means converting format 2 workflows into standard Galaxy workflow JSON for consumption for the rest of this module. In the future we will want to be a lot more precise about this - preserve the original description along side the data model and apply updates in a way that largely preserves YAML structure so workflows can be extracted.

refactor(trans, stored_workflow, refactor_request)[source]
update_workflow_from_raw_description(trans, stored_workflow, raw_workflow_description, workflow_update_options)[source]
workflow_to_dict(trans, stored, style='export', version=None, history=None)[source]

Export the workflow contents to a dictionary ready for JSON-ification and to be sent out via API for instance. There are three styles of export allowed ‘export’, ‘instance’, and ‘editor’. The Galaxy team will do its best to preserve the backward compatibility of the ‘export’ style - this is the export method meant to be portable across Galaxy instances and over time. The ‘editor’ style is subject to rapid and unannounced changes. The ‘instance’ export option describes the workflow in a context more tied to the current Galaxy instance and includes fields like ‘url’ and ‘url’ and actual unencoded step ids instead of ‘order_index’.

class galaxy.managers.workflows.WorkflowCreateOptions[source]

Bases: galaxy.managers.workflows.WorkflowStateResolutionOptions

install_options
is_importable
class galaxy.managers.workflows.WorkflowStateResolutionOptions[source]

Bases: pydantic.main.BaseModel

class galaxy.managers.workflows.WorkflowUpdateOptions[source]

Bases: galaxy.managers.workflows.WorkflowStateResolutionOptions

class galaxy.managers.workflows.WorkflowsManager(app: galaxy.structured_app.MinimalManagerApp)[source]

Bases: object

Handle CRUD type operations related to workflows. More interesting stuff regarding workflow execution, step sorting, etc… can be found in the galaxy.workflow module.

__init__(app: galaxy.structured_app.MinimalManagerApp)[source]
attach_stored_workflow(trans, workflow)[source]

Attach and return stored workflow if possible.

build_invocations_query(trans, stored_workflow_id=None, history_id=None, job_id=None, user_id=None, include_terminal=True, limit=None, offset=None, sort_by=None, sort_desc=None)[source]

Get invocations owned by the current user.

cancel_invocation(trans, decoded_invocation_id)[source]
check_security(trans, has_workflow, check_ownership=True, check_accessible=True)[source]

check accessibility or ownership of workflows, storedworkflows, and workflowinvocations. Throw an exception or returns True if user has needed level of access.

get_invocation(trans, decoded_invocation_id, eager=False)[source]
get_invocation_report(trans, invocation_id, **kwd)[source]
get_invocation_step(trans, decoded_workflow_invocation_step_id)[source]
get_owned_workflow(trans, encoded_workflow_id)[source]

Get a workflow (non-stored) from a encoded workflow id and make sure it accessible to the user.

get_stored_accessible_workflow(trans, workflow_id, by_stored_id=True)[source]

Get a stored workflow from a encoded stored workflow id and make sure it accessible to the user.

get_stored_workflow(trans, workflow_id, by_stored_id=True)[source]

Use a supplied ID (UUID or encoded stored workflow ID) to find a workflow.

serialize_workflow_invocation(invocation, **kwd)[source]
serialize_workflow_invocations(invocations, **kwd)[source]
update_invocation_step(trans, decoded_workflow_invocation_step_id, action)[source]