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.AnnotatableManagerMixin

Bases: object

annotation_assoc: type

class of AnnotationAssociation (e.g. HistoryAnnotationAssociation)

abstract session() → scoped_session

annotation(item) → str | None

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

annotate(item, annotation, user=None, flush=True)

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

class galaxy.managers.annotatable.AnnotatableSerializerMixin

Bases: object

serializers: Dict[str, Serializer]

add_serializers()

serialize_annotation(item, key, user=None, **context)

Get and serialize an item’s annotation.

class galaxy.managers.annotatable.AnnotatableDeserializerMixin

Bases: object

deserializers: Dict[str, Deserializer]

add_deserializers()

deserialize_annotation(item, key, val, user=None, **context)

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

class galaxy.managers.annotatable.AnnotatableFilterMixin

Bases: object

fn_filter_parsers: Dict[str, Any]

filter_annotation_contains(item, val: str) → bool

Test whether val is in the owner’s annotation.

galaxy.managers.api_keys module

class galaxy.managers.api_keys.IsUserModel(*args, **kwargs)

Bases: Protocol

id: int

__init__(*args, **kwargs)

class galaxy.managers.api_keys.ApiKeyManager(app: BasicSharedApp)

Bases: object

__init__(app: BasicSharedApp)

get_api_key(user: IsUserModel)

create_api_key(user: IsUserModel)

get_or_create_api_key(user: IsUserModel) → str

delete_api_key(user: IsUserModel) → None

Marks the current user API key as deleted.

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.ParsedFilter(filter_type, filter, case_insensitive)

Bases: tuple

filter_type: str

Alias for field number 0

filter: Any

Alias for field number 1

case_insensitive: bool

Alias for field number 2

galaxy.managers.base.parsed_filter

alias of ParsedFilter

galaxy.managers.base.security_check(trans, item, check_ownership=False, check_accessible=False)

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.base.get_class(class_name)

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

galaxy.managers.base.decode_id(app: BasicSharedApp, id: Any, kind: str | None = None) → int

galaxy.managers.base.decode_with_security(security: IdEncodingHelper, id: Any, kind: str | None = None)

galaxy.managers.base.encode_with_security(security: IdEncodingHelper, id: Any, kind: str | None = None)

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

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.

class galaxy.managers.base.ModelManager(app: BasicSharedApp)

Bases: Generic[U]

Base class for all model/resource managers.

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

model_class: Type[U]

foreign_key_name: str

__init__(app: BasicSharedApp)

app: BasicSharedApp

session() → scoped_session

query(eagerloads: bool = True, filters=None, order_by=None, limit: int | None = None, offset: int | None = None) → Query

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

Set eagerloads to False to disable them for this query.

one(**kwargs) → U

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

by_id(id: int) → U

Gets a model by primary id.

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

Returns all objects matching the given filters

count(filters=None, **kwargs) → int

Returns the number of objects matching the given filters.

If the filters include functional filters, this function will raise an exception as they might cause performance issues.

by_ids(ids, filters=None, **kwargs)

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

create(flush: bool = True, *args: Any, **kwargs: Any) → U

Generically create a new model.

copy(item, **kwargs) → U

Clone or copy an item.

update(item: U, new_values: Dict[str, Any], flush: bool = True, **kwargs) → U

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

..note: NO validation or deserialization occurs here.

associate(associate_with, item, foreign_key_name=None)

Generically associate item with associate_with based on foreign_key_name.

query_associated(associated_model_class, item, foreign_key_name=None)

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

class galaxy.managers.base.HasAModelManager(app: MinimalManagerApp, manager=None, **kwargs)

Bases: Generic[T]

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.

model_manager_class: Type[T]

the class used to create this serializer’s generically accessible model_manager

__init__(app: MinimalManagerApp, manager=None, **kwargs)

app: MinimalManagerApp

property manager: T

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

exception galaxy.managers.base.ModelSerializingError(err_msg: str | None = None, type='info', **extra_error_info)

Bases: InternalServerError

Thrown when request model values can’t be serialized

exception galaxy.managers.base.ModelDeserializingError(err_msg: str | None = None, type='info', **extra_error_info)

Bases: ObjectAttributeInvalidException

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

exception galaxy.managers.base.SkipAttribute

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.Serializer(*args, **kwargs)

Bases: Protocol

__init__(*args, **kwargs)

galaxy.managers.base.url_for_not_available(*args, **kwargs)

class galaxy.managers.base.ModelSerializer(app: MinimalManagerApp, **kwargs)

Bases: HasAModelManager[T]

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: MinimalManagerApp, **kwargs)

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

views: Dict[str, List[str]]

default_view: str | None

static url_for(*args, context=None, **kwargs)

add_serializers()

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

add_view(view_name, key_list, include_keys_from=None)

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.

serialize(item, keys, **context)

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.

skip(msg='skipped')

To be called from inside a serializer to skip it.

Handy for config checks, information hiding, etc.

default_serializer(item, key, **context)

Serialize the item’s attribute named key.

serialize_date(item: Any, key: str, **context)

Serialize a date attribute of item.

serialize_id(item: Any, key: str, encode_id=True, **context)

Serialize an id attribute of item.

serialize_type_id(item: Any, key: str, **context)

Serialize an type-id for item.

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

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

class galaxy.managers.base.ModelValidator

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 matches_type(key: str, val: Any, types: type | Tuple[type | Tuple[Any, ...], ...])

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

Raises:

exceptions.RequestParameterInvalidException – if not an instance.

static basestring(key: str, val: Any) → str

static bool(key: str, val: Any) → bool

static nullable_basestring(key: str, val: Any) → str

Must be a basestring or None.

static int_range(key: str, val: Any, min: int | None = None, max: int | None = None) → int

Must be a int between min and max.

static basestring_list(key: str, val: Any) → List[str]

Must be a list of basestrings.

static genome_build(key: str, val: Any) → str

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.

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

Bases: Protocol

__init__(*args, **kwargs)

class galaxy.managers.base.ModelDeserializer(app: MinimalManagerApp, **kwargs)

Bases: HasAModelManager[T]

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

validate = <galaxy.managers.base.ModelValidator object>

app: MinimalManagerApp

__init__(app: MinimalManagerApp, **kwargs)

Set up deserializers and validator.

add_deserializers()

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

deserialize(item, data, flush=True, **context)

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

default_deserializer(item, key, val, **context)

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

deserialize_basestring(item, key, val, convert_none_to_empty=False, **context)

deserialize_bool(item, key, val, **context)

deserialize_int(item, key, val, min=None, max=None, **context)

deserialize_genome_build(item, key, val, **context)

Make sure val is a valid dbkey and assign it.

class galaxy.managers.base.ModelFilterParser(app: MinimalManagerApp, **kwargs)

Bases: 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.

model_class: Type[object]

parsed_filter

alias of ParsedFilter

__init__(app: MinimalManagerApp, **kwargs)

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

date_string_re

regex for testing/dicing iso8601 date strings, with optional time and ms, but allowing only UTC timezone

orm_filter_parsers: Dict[str, None | Dict[str, Any] | Callable]

fn_filter_parsers: Dict[str, Any]

dictionary containing parsing data for functional filters - applied after a query is made

build_filter_params(query_params: ValueFilterQueryParams, filter_attr_key: str = 'q', filter_value_key: str = 'qv', attr_op_split_char: str = '-') → List[Tuple[str, str, str]]

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

parse_query_filters(query_filters: ValueFilterQueryParams)

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

parse_filters(filter_tuple_list)

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

parse_filter(attr, op, val)

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.

UNDERSCORED_OPS = ('lt', 'le', 'eq', 'ne', 'ge', 'gt')

these are the easier/shorter string equivalents to the python operator fn names that need ‘__’ around them

string_standard_ops(key)

parse_id_list(id_list_string, sep=',')

Split id_list_string at sep and decode as ids.

parse_int_list(int_list_string, sep=',')

Split int_list_string at sep and parse as ints.

parse_date(date_string)

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).

contains_non_orm_filter(filters: List[ParsedFilter]) → bool

Whether the list of filters contains any non-orm filter.

galaxy.managers.base.parse_bool(bool_string: str | bool) → bool

Parse a boolean from a string.

galaxy.managers.base.raise_filter_err(attr, op, val, msg)

class galaxy.managers.base.SortableManager

Bases: object

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

parse_order_by(order_by_string, default=None)

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

class galaxy.managers.base.StorageCleanerManager(*args, **kwargs)

Bases: Protocol

Interface for monitoring storage usage and managing deletion/purging of objects that consume user’s storage space.

sort_map: Dict[StoredItemOrderBy, Any]

__init__(*args, **kwargs)

get_discarded_summary(user: User) → CleanableItemsSummary

Returns information with the total storage space taken by discarded items for the given user.

Discarded items are those that are deleted but not purged yet.

get_discarded(user: User, offset: int | None, limit: int | None, order: StoredItemOrderBy | None) → List[StoredItem]

Returns a paginated list of items deleted by the given user that are not yet purged.

get_archived_summary(user: User) → CleanableItemsSummary

Returns information with the total storage space taken by archived items for the given user.

Archived items are those that are not currently active. Some archived items may be purged already, but this method does not return information about those.

get_archived(user: User, offset: int | None, limit: int | None, order: StoredItemOrderBy | None) → List[StoredItem]

Returns a paginated list of items archived by the given user that are not yet purged.

cleanup_items(user: User, item_ids: Set[int]) → StorageItemsCleanupResult

Purges the given list of items by ID. The items must be owned by the user.

galaxy.managers.base.combine_lists(listA: Any, listB: Any) → List

Combine two lists into a single list.

Arguments can be None, non-lists, or lists. If an argument is None, it will not be included in the returned list. If both arguments are None, an empty list will be returned.

galaxy.managers.citations module

class galaxy.managers.citations.CitationsManager(app: BasicSharedApp)

Bases: object

__init__(app: BasicSharedApp) → None

citations_for_tool(tool)

citations_for_tool_ids(tool_ids)

parse_citation(citation_model: Citation) → BibtexCitation | DoiCitation | None

class galaxy.managers.citations.DoiCache(config)

Bases: object

__init__(config)

get_bibtex(doi)

galaxy.managers.citations.parse_citation(citation_model: Citation, citation_manager) → BibtexCitation | DoiCitation | None

Parse an abstract citation entry from the specified XML element.

class galaxy.managers.citations.CitationCollection

Bases: object

__init__()

add(new_citation)

class galaxy.managers.citations.BaseCitation

Bases: object

to_dict(citation_format)

equals(other_citation)

has_doi()

class galaxy.managers.citations.BibtexCitation(citation_model: Citation, citation_manager: CitationsManager)

Bases: BaseCitation

__init__(citation_model: Citation, citation_manager: CitationsManager)

to_bibtex() → str

class galaxy.managers.citations.DoiCitation(citation_model: Citation, citation_manager: CitationsManager)

Bases: BaseCitation

__init__(citation_model: Citation, citation_manager: CitationsManager)

has_doi()

doi()

to_bibtex() → str

galaxy.managers.cloud module

galaxy.managers.cloudauthzs module

galaxy.managers.collections module

class galaxy.managers.collections.DatasetCollectionManager(model: GalaxyModelMapping, security: IdEncodingHelper, hda_manager: HDAManager, history_manager: HistoryManager, ldda_manager: LDDAManager, short_term_storage_monitor: ShortTermStorageMonitor)

Bases: object

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

ELEMENTS_UNINITIALIZED = <object object>

__init__(model: GalaxyModelMapping, security: IdEncodingHelper, hda_manager: HDAManager, history_manager: HistoryManager, ldda_manager: LDDAManager, short_term_storage_monitor: ShortTermStorageMonitor)

precreate_dataset_collection_instance(trans: ProvidesHistoryContext, parent, name, structure, implicit_inputs=None, implicit_output_name=None, tags=None, completed_collection=None)

precreate_dataset_collection(structure, allow_uninitialized_element=True, completed_collection=None, implicit_output_name=None)

create(trans: ProvidesHistoryContext, 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)

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

create_dataset_collection(trans: ProvidesHistoryContext, collection_type, element_identifiers=None, elements=None, hide_source_items=None, copy_elements=False, history=None)

get_converters_for_collection(trans: ProvidesHistoryContext, id, datatypes_registry: Registry, instance_type='history')

collection_builder_for(dataset_collection)

delete(trans: ProvidesHistoryContext, instance_type, id, recursive=False, purge=False)

update(trans: ProvidesHistoryContext, instance_type, id, payload)

copy(trans: ProvidesHistoryContext, parent, source, encoded_source_id, copy_elements=False, dataset_instance_attributes=None)

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

history_dataset_collections(history, query)

match_collections(collections_to_match)

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.

get_dataset_collection_instance(trans: ProvidesHistoryContext, instance_type: typing_extensions.Literal[history], id, **kwds: Any) → HistoryDatasetCollectionAssociation

get_dataset_collection_instance(trans: ProvidesHistoryContext, instance_type: typing_extensions.Literal[library], id, **kwds: Any) → LibraryDatasetCollectionAssociation

get_dataset_collection(trans, encoded_id)

apply_rules(hdca, rule_set, handle_dataset)

get_collection_contents(trans: ProvidesAppContext, parent_id, limit=None, offset=None)

Find first level of collection contents by containing collection parent_id

write_dataset_collection(request: PrepareDatasetCollectionDownload)

galaxy.managers.collections_util module

galaxy.managers.collections_util.api_payload_to_create_params(payload)

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)

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.ConfigurationManager(app: StructuredApp)

Bases: object

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

__init__(app: StructuredApp)

get_configuration(trans: ProvidesUserContext, serialization_params: SerializationParams) → Dict[str, Any]

version() → Dict[str, Any]

decode_id(encoded_id: str) → Dict[str, int]

encode_id(decoded_id: int) → Dict[str, str]

tool_lineages() → List[Dict[str, Dict]]

dynamic_tool_confs() → List[Dict[str, str]]

reload_toolbox()

class galaxy.managers.configuration.ConfigSerializer(app)

Bases: ModelSerializer

Configuration (galaxy.ini) settings viewable by all users

__init__(app)

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

default_view: str | None

default_serializer(item, key, **context)

Serialize the item’s attribute named key.

add_serializers()

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

views: Dict[str, List[str]]

serializable_keyset: Set[str]

serializers: Dict[str, Serializer]

model_manager_class: Type[T]

the class used to create this serializer’s generically accessible model_manager

app: MinimalManagerApp

class galaxy.managers.configuration.AdminConfigSerializer(app)

Bases: ConfigSerializer

Configuration attributes viewable only by admin users

add_serializers()

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

default_view: str | None

views: Dict[str, List[str]]

serializable_keyset: Set[str]

serializers: Dict[str, Serializer]

model_manager_class: Type[T]

the class used to create this serializer’s generically accessible model_manager

app: MinimalManagerApp

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

Bases: object

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

Mixed in class must provide app property.

abstract property app: MinimalManagerApp

Provide access to the Galaxy app object.

abstract property url_builder: Callable[[...], str] | None

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).

property security: IdEncodingHelper

Provide access to Galaxy app’s id encoding helper.

Return type:

IdEncodingHelper

log_action(user=None, action=None, context=None, params=None)

Application-level logging of user actions.

log_event(message, tool_id=None, **kwargs)

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

property sa_session: galaxy_scoped_session

Provide access to Galaxy’s SQLAlchemy session.

Return type:

galaxy.model.scoped_session.galaxy_scoped_session

get_toolbox()

Returns the application toolbox.

Return type:

galaxy.tools.ToolBox

property model: ModelMapping

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

property install_model: ModelMapping

Provide access to Galaxy’s install mapping.

Comments on the model property apply here also.

class galaxy.managers.context.ProvidesUserContext

Bases: ProvidesAppContext

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

Mixed in class must provide user and app properties.

galaxy_session: GalaxySession | None = None

set_cache_value(args: Tuple[str, ...], value: Any)

get_cache_value(args: Tuple[str, ...], default: Any | None = None) → Any

property tag_handler

property async_request_user: RequestUser

abstract property user

Provide access to the user object.

property user_vault

Provide access to a user’s personal vault.

get_user() → User | None

property anonymous: bool

get_current_user_roles() → List[Role]

property user_is_admin: bool

property user_is_bootstrap_admin: bool

Master key provided so there is no real user

property user_can_do_run_as: bool

property user_is_active: bool

check_user_activation()

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

property user_ftp_dir: str | None

class galaxy.managers.context.ProvidesHistoryContext

Bases: ProvidesUserContext

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

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

abstract property history: History | None

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

Return type:

Optional[galaxy.model.History]

db_dataset_for(dbkey) → HistoryDatasetAssociation | None

Optionally return the db_file dataset associated/needed by dataset.

property db_builds

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

galaxy.managers.context.get_hdas(session, history_id, states)

galaxy.managers.datasets module

Manager and Serializer for Datasets.

class galaxy.managers.datasets.DatasetManager(app: MinimalManagerApp)

Bases: ModelManager[Dataset], AccessibleManagerMixin, PurgableManagerMixin

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

model_class

alias of Dataset

foreign_key_name: str = 'dataset'

app: MinimalManagerApp

__init__(app: MinimalManagerApp)

copy(item, **kwargs)

Clone or copy an item.

purge(item, flush=True, **kwargs)

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

Raises:

exceptions.ConfigDoesNotAllowException – if the instance doesn’t allow

purge_datasets(request: PurgeDatasetsTaskRequest)

Caution: any additional security checks must be done before executing this action.

Completely removes a set of object_store/files associated with the datasets from storage and marks them as purged. They might not be removed if there are still un-purged associations to the dataset.

error_unless_dataset_purge_allowed(msg=None)

is_accessible(item: Any, user: User | None, **kwargs) → bool

Is this dataset readable/viewable to user?

has_access_permission(dataset, user)

Whether the user has role-based access to the dataset.

update_object_store_id(trans, dataset, object_store_id: str)

compute_hash(request: ComputeDatasetHashTaskRequest)

class galaxy.managers.datasets.DatasetRBACPermissions(app)

Bases: object

__init__(app)

available_roles(trans, dataset, controller='root')

get(dataset, flush=True)

set(dataset, manage_roles, access_roles, flush=True)

set_public_with_single_manager(dataset, user, flush=True)

set_private_to_one_user(dataset, user, flush=True)

class galaxy.managers.datasets.DatasetSerializer(app: MinimalManagerApp, user_manager: UserManager)

Bases: ModelSerializer[DatasetManager], PurgableSerializerMixin

model_manager_class

alias of DatasetManager

__init__(app: MinimalManagerApp, user_manager: UserManager)

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

default_view: str | None

add_serializers()

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

serialize_file_name(item, key, user=None, **context)

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

serialize_extra_files_path(item, key, user=None, **context)

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

serialize_permissions(item, key, user=None, **context)

views: Dict[str, List[str]]

serializable_keyset: Set[str]

serializers: Dict[str, Serializer]

app: MinimalManagerApp

class galaxy.managers.datasets.DatasetAssociationManager(app: MinimalManagerApp)

Bases: ModelManager[DatasetInstance], AccessibleManagerMixin, OwnableManagerMixin, PurgableManagerMixin, Generic[U]

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

app: MinimalManagerApp

__init__(app: MinimalManagerApp)

is_accessible(item: U, user: User | None, **kwargs: Any) → bool

Is this DA accessible to user?

delete(item: U, flush: bool = True, stop_job: bool = False, **kwargs)

Marks this dataset association as deleted. If stop_job is True, will stop the creating job if all other outputs are deleted.

purge(item: U, flush=True, **kwargs)

Purge this DatasetInstance and the dataset underlying it.

by_user(user)

creating_job(dataset_assoc: U)

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

stop_creating_job(dataset_assoc: U, flush=False)

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

is_composite(dataset_assoc: U)

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

Note

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

extra_files(dataset_assoc: U)

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

serialize_dataset_association_roles(dataset_assoc: U)

ensure_dataset_on_disk(trans, dataset: U)

ensure_can_change_datatype(dataset: U, raiseException: bool = True) → bool

ensure_can_set_metadata(dataset: U, raiseException: bool = True) → bool

detect_datatype(trans, dataset_assoc: U)

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

set_metadata(trans, dataset_assoc: U, overwrite: bool = False, validate: bool = True) → None

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

overwrite_metadata(data)

update_permissions(trans, dataset_assoc: U, **kwd)

class galaxy.managers.datasets.DatasetAssociationSerializer(app)

Bases: _UnflattenedMetadataDatasetAssociationSerializer[T]

add_serializers()

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

serialize(item, keys, **context)

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

default_view: str | None

views: Dict[str, List[str]]

serializable_keyset: Set[str]

serializers: Dict[str, Serializer]

model_manager_class: Type[T]

the class used to create this serializer’s generically accessible model_manager

app: MinimalManagerApp

class galaxy.managers.datasets.DatasetAssociationDeserializer(app: MinimalManagerApp, **kwargs)

Bases: ModelDeserializer, PurgableDeserializerMixin

add_deserializers()

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

deserialize_metadata(dataset_assoc, metadata_key, metadata_dict, **context)

deserialize_metadatum(dataset_assoc, key, val, **context)

deserialize_datatype(item, key, val, **context)

app: MinimalManagerApp

deserializable_keyset: Set[str]

model_manager_class: Type[T]

the class used to create this serializer’s generically accessible model_manager

class galaxy.managers.datasets.DatasetAssociationFilterParser(app: MinimalManagerApp, **kwargs)

Bases: ModelFilterParser, PurgableFiltersMixin

eq_datatype(dataset_assoc, class_str)

Is the dataset_assoc datatype equal to the registered datatype class_str?

isinstance_datatype(dataset_assoc, class_strs)

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

model_class: Type[object]

orm_filter_parsers: Dict[str, None | Dict[str, Any] | Callable]

fn_filter_parsers: Dict[str, Any]

dictionary containing parsing data for functional filters - applied after a query is made

model_manager_class: Type[T]

the class used to create this serializer’s generically accessible model_manager

app: MinimalManagerApp

galaxy.managers.datasets.get_dataset_hash(session, dataset_id, hash_function, extra_files_path)

galaxy.managers.datatypes module

class galaxy.managers.datatypes.DatatypeConverterList(root: RootModelRootType = PydanticUndefined)

Bases: RootModel

root: List[DatatypeConverter]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'root': FieldInfo(annotation=List[galaxy.datatypes._schema.DatatypeConverter], required=False, default=[], title='List of data type converters')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class galaxy.managers.datatypes.DatatypeDetails(*, extension: str, description: str | None, description_url: Url[Url] | None, display_in_upload: bool = False, composite_files: List[CompositeFileInfo] | None = None, upload_warning: str | None = None)

Bases: BaseModel

extension: str

description: str | None

description_url: Url[Url] | None

display_in_upload: bool

composite_files: List[CompositeFileInfo] | None

upload_warning: str | None

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'composite_files': FieldInfo(annotation=Union[List[galaxy.datatypes._schema.CompositeFileInfo], NoneType], required=False, default=None, title='Composite files', description='A collection of files composing this data type'), 'description': FieldInfo(annotation=Union[str, NoneType], required=True, title='Description', description='A summary description for this data type'), 'description_url': FieldInfo(annotation=Union[Annotated[pydantic_core._pydantic_core.Url, UrlConstraints(max_length=2083, allowed_schemes=['http', 'https'], host_required=None, default_host=None, default_port=None, default_path=None)], NoneType], required=True, title='Description URL', description='The URL to a detailed description for this datatype', examples=['https://wiki.galaxyproject.org/Learn/Datatypes#Bed']), 'display_in_upload': FieldInfo(annotation=bool, required=False, default=False, title='Display in upload', description='If True, the associated file extension will be displayed in the `File Format` select list in the `Upload File from your computer` tool in the `Get Data` tool section of the tool panel'), 'extension': FieldInfo(annotation=str, required=True, title='Extension', description='The data type’s Dataset file extension', examples=['bed']), 'upload_warning': FieldInfo(annotation=Union[str, NoneType], required=False, default=None, title='Upload warning', description='End-user information regarding potential pitfalls with this upload type.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class galaxy.managers.datatypes.DatatypesCombinedMap(*, datatypes: List[str], datatypes_mapping: DatatypesMap)

Bases: BaseModel

datatypes: List[str]

datatypes_mapping: DatatypesMap

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'datatypes': FieldInfo(annotation=List[str], required=True, title='Datatypes', description='List of datatypes extensions'), 'datatypes_mapping': FieldInfo(annotation=DatatypesMap, required=True, title='Datatypes Mapping', description="Dictionaries for mapping datatype's extensions/classes with their implementation classes")}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class galaxy.managers.datatypes.DatatypesEDAMDetailsDict(root: RootModelRootType = PydanticUndefined)

Bases: RootModel

root: Dict[str, DatatypeEDAMDetails]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'root': FieldInfo(annotation=Dict[str, galaxy.datatypes._schema.DatatypeEDAMDetails], required=False, default={}, title='Dict of EDAM details for formats')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class galaxy.managers.datatypes.DatatypesMap(*, ext_to_class_name: Dict[str, str], class_to_classes: Dict[str, Dict[str, bool]])

Bases: BaseModel

ext_to_class_name: Dict[str, str]

class_to_classes: Dict[str, Dict[str, bool]]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'class_to_classes': FieldInfo(annotation=Dict[str, Dict[str, bool]], required=True, title='Classes Map', description="Dictionary mapping datatype's classes with their base classes"), 'ext_to_class_name': FieldInfo(annotation=Dict[str, str], required=True, title='Extension Map', description="Dictionary mapping datatype's extensions with implementation classes")}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

galaxy.managers.datatypes.view_index(datatypes_registry: Registry, extension_only: bool | None = True, upload_only: bool | None = True) → List[DatatypeDetails] | List[str]

galaxy.managers.datatypes.view_mapping(datatypes_registry: Registry) → DatatypesMap

galaxy.managers.datatypes.view_types_and_mapping(datatypes_registry: Registry, extension_only: bool | None = True, upload_only: bool | None = True) → DatatypesCombinedMap

galaxy.managers.datatypes.view_sniffers(datatypes_registry: Registry) → List[str]

galaxy.managers.datatypes.view_converters(datatypes_registry: Registry) → DatatypeConverterList

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.DeletableManagerMixin

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)

Mark as deleted and return.

undelete(item, flush=True, **kwargs)

Mark as not deleted and return.

class galaxy.managers.deletable.DeletableSerializerMixin

Bases: object

serializable_keyset: Set[str]

add_serializers()

class galaxy.managers.deletable.DeletableDeserializerMixin

Bases: object

deserializers: Dict[str, Deserializer]

add_deserializers()

deserialize_deleted(item, key, val, **context)

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

class galaxy.managers.deletable.DeletableFiltersMixin

Bases: object

orm_filter_parsers: Dict[str, None | Dict[str, Any] | Callable]

class galaxy.managers.deletable.PurgableManagerMixin

Bases: 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)

Mark as purged and return.

Override this in subclasses to do the additional resource removal.

class galaxy.managers.deletable.PurgableSerializerMixin

Bases: DeletableSerializerMixin

serializable_keyset: Set[str]

add_serializers()

class galaxy.managers.deletable.PurgableDeserializerMixin

Bases: DeletableDeserializerMixin

deserializers: Dict[str, Deserializer] = {}

add_deserializers()

deserialize_purged(item, key, val, **context)

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

class galaxy.managers.deletable.PurgableFiltersMixin

Bases: DeletableFiltersMixin

orm_filter_parsers: Dict[str, None | Dict[str, Any] | Callable]

galaxy.managers.display_applications module

class galaxy.managers.display_applications.Link(*, name: str)

Bases: BaseModel

name: str

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'name': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class galaxy.managers.display_applications.DisplayApplication(*, id: str, name: str, version: str, filename_: str, links: List[Link])

Bases: BaseModel

id: str

name: str

version: str

filename_: str

links: List[Link]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'filename_': FieldInfo(annotation=str, required=True), 'id': FieldInfo(annotation=str, required=True), 'links': FieldInfo(annotation=List[galaxy.managers.display_applications.Link], required=True), 'name': FieldInfo(annotation=str, required=True), 'version': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class galaxy.managers.display_applications.ReloadFeedback(*, message: str, reloaded: List[str | None], failed: List[str | None])

Bases: BaseModel

message: str

reloaded: List[str | None]

failed: List[str | None]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'failed': FieldInfo(annotation=List[Union[str, NoneType]], required=True), 'message': FieldInfo(annotation=str, required=True), 'reloaded': FieldInfo(annotation=List[Union[str, NoneType]], required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class galaxy.managers.display_applications.DisplayApplicationsManager(app: StructuredApp)

Bases: object

Interface/service object for sharing logic between controllers.

__init__(app: StructuredApp)

property datatypes_registry: Registry

index() → List[DisplayApplication]

Returns the list of display applications.

Returns:

list of available display applications

Return type:

list

reload(ids: List[str]) → ReloadFeedback

Reloads the list of display applications.

Parameters:

ids (list) – list containing ids of display to be reloaded

galaxy.managers.executables module

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

galaxy.managers.executables.artifact_class(trans, as_dict: Dict[str, Any], allow_in_directory: str | None = None)

galaxy.managers.folders module

Manager and Serializer for Library Folders.

class galaxy.managers.folders.SecurityParams(user_role_ids: List[Role], security_agent: RBACAgent, is_admin: bool)

Bases: object

Contains security data bundled for reusability.

user_role_ids: List[Role]

security_agent: RBACAgent

is_admin: bool

__init__(user_role_ids: List[Role], security_agent: RBACAgent, is_admin: bool) → None

class galaxy.managers.folders.FolderManager

Bases: object

Interface/service object for interacting with folders.

get(trans, decoded_folder_id: int, check_manageable: bool = False, check_accessible: bool = True)

Get the folder from the DB.

Parameters:

• decoded_folder_id – decoded folder id

• check_manageable – flag whether the check that user can manage item

• check_accessible – flag whether to check that user can access item

Returns:

the requested folder

Return type:

LibraryFolder

Raises:

InconsistentDatabase, RequestParameterInvalidException, InternalServerError

secure(trans, folder, check_manageable=True, check_accessible=True)

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

check_modifyable(trans, folder)

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

Returns:

the original folder

Return type:

LibraryFolder

Raises:

AuthenticationRequired, InsufficientPermissionsException

check_manageable(trans, folder)

Check whether the user can manage the folder.

Returns:

the original folder

Return type:

LibraryFolder

Raises:

AuthenticationRequired, InsufficientPermissionsException

check_accessible(trans, folder)

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

get_folder_dict(trans, folder)

Return folder data in the form of a dictionary.

Parameters:

folder (LibraryFolder) – folder item

Returns:

dict with data about the folder

Return type:

dictionary

create(trans, parent_folder_id, new_folder_name, new_folder_description='')

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

update(trans, folder, name=None, description=None)

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

delete(trans, folder, undelete=False)

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_current_roles(trans, folder)

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

can_add_item(trans, folder)

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

cut_the_prefix(encoded_folder_id)

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)

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

cut_and_decode(trans, encoded_folder_id)

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

get_contents(trans, folder: LibraryFolder, payload: LibraryFolderContentsIndexQueryPayload) → Tuple[List[LibraryFolder | LibraryDataset], int]

Retrieves the contents of the given folder that match the provided filters and pagination parameters. Returns a tuple with the list of paginated contents and the total number of items contained in the folder.

build_folder_path(sa_session: galaxy_scoped_session, folder: LibraryFolder) → List[Tuple[int, str | None]]

Returns the folder path from root to the given folder.

The path items are tuples with the name and id of each folder for breadcrumb building purposes.

galaxy.managers.folders.get_folder(session, folder_id)

galaxy.managers.folders.get_count(session, statement)

galaxy.managers.genomes module

class galaxy.managers.genomes.GenomesManager(app: StructuredApp)

Bases: object

__init__(app: StructuredApp)

get_dbkeys(user: User | None, chrom_info: bool) → List[List[str]]

is_registered_dbkey(dbkey: str, user: User | None) → bool

get_genome(trans: ProvidesUserContext, id: str, num: int, chrom: str, low: int, high: int, reference: bool) → Any

get_sequence(trans: ProvidesUserContext, id: str, chrom: str, low: int, high: int) → Any

get_indexes(id: str, index_type: str) → Any

class galaxy.managers.genomes.GenomeFilterMixin

Bases: object

app: MinimalManagerApp

orm_filter_parsers: OrmFilterParsersType

valid_ops = ('eq', 'contains', 'has')

create_genome_filter(attr, op, val)

galaxy.managers.group_roles module

class galaxy.managers.group_roles.GroupRolesManager(app: MinimalManagerApp)

Bases: object

Interface/service object shared by controllers for interacting with group roles.

__init__(app: MinimalManagerApp) → None

index(trans: ProvidesAppContext, group_id: int) → List[GroupRoleAssociation]

Returns a collection roles associated with the given group.

show(trans: ProvidesAppContext, role_id: int, group_id: int) → Role

Returns information about a group role.

update(trans: ProvidesAppContext, role_id: int, group_id: int) → Role

Adds a role to a group if it is not already associated.

delete(trans: ProvidesAppContext, role_id: int, group_id: int) → Role

Removes a role from a group.

galaxy.managers.group_roles.get_group_role(session: galaxy_scoped_session, group, role) → GroupRoleAssociation | None

galaxy.managers.group_users module

class galaxy.managers.group_users.GroupUsersManager(app: MinimalManagerApp)

Bases: object

Interface/service object shared by controllers for interacting with group users.

__init__(app: MinimalManagerApp) → None

index(trans: ProvidesAppContext, group_id: int) → List[User]

Returns a collection (list) with some information about users associated with the given group.

show(trans: ProvidesAppContext, user_id: int, group_id: int) → User

Returns information about a group user.

update(trans: ProvidesAppContext, user_id: int, group_id: int) → User

Adds a user to a group.

delete(trans: ProvidesAppContext, user_id: int, group_id: int) → User

Removes a user from a group.

galaxy.managers.group_users.get_group_user(session: galaxy_scoped_session, user, group) → UserGroupAssociation | None

galaxy.managers.groups module

class galaxy.managers.groups.GroupsManager(app: MinimalManagerApp)

Bases: object

Interface/service object shared by controllers for interacting with groups.

__init__(app: MinimalManagerApp) → None

index(trans: ProvidesAppContext)

Displays a collection (list) of groups.

create(trans: ProvidesAppContext, payload: GroupCreatePayload)

Creates a new group.

show(trans: ProvidesAppContext, group_id: int)

Displays information about a group.

update(trans: ProvidesAppContext, group_id: int, payload: GroupUpdatePayload)

Modifies a group.

delete(trans: ProvidesAppContext, group_id: int)

purge(trans: ProvidesAppContext, group_id: int)

undelete(trans: ProvidesAppContext, group_id: int)

galaxy.managers.groups.get_group_by_name(session: galaxy_scoped_session, name: str)

galaxy.managers.groups.get_not_deleted_groups(session: galaxy_scoped_session)

galaxy.managers.hdas module

Manager and Serializer for HDAs.

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

exception galaxy.managers.hdas.HistoryDatasetAssociationNoHistoryException

Bases: Exception

class galaxy.managers.hdas.HDAManager(app: MinimalManagerApp, user_manager: UserManager, ldda_manager: LDDAManager)

Bases: DatasetAssociationManager[HistoryDatasetAssociation], OwnableManagerMixin, AnnotatableManagerMixin

Interface/service object for interacting with HDAs.

model_class

alias of HistoryDatasetAssociation

foreign_key_name: str = 'history_dataset_association'

tag_assoc

alias of HistoryDatasetAssociationTagAssociation

annotation_assoc

alias of HistoryDatasetAssociationAnnotationAssociation

app: MinimalManagerApp

__init__(app: MinimalManagerApp, user_manager: UserManager, ldda_manager: LDDAManager)

Set up and initialize other managers needed by hdas.

get_owned_ids(object_ids, history=None)

Get owned IDs.

is_owner(item, user: User | None, current_history=None, **kwargs: Any) → bool

Use history to see if current user owns HDA.

create(flush: bool = True, history=None, dataset=None, *args: Any, **kwargs: Any) → HistoryDatasetAssociation

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.

materialize(request: MaterializeDatasetInstanceTaskRequest, in_place: bool = False) → bool

copy(item: Any, history=None, hide_copy: bool = False, flush: bool = True, **kwargs: Any) → HistoryDatasetAssociation

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

purge(item, flush=True, **kwargs)

Purge this DatasetInstance and the dataset underlying it.

error_if_uploading(hda)

Raise error if HDA is still uploading.

has_been_resubmitted(hda)

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

data_conversion_status(hda)

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

text_data(hda, preview=True)

Get data from text file, truncating if necessary.

annotation(hda)

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

galaxy.managers.hdas.dereference_input(trans: ProvidesHistoryContext, data_request: DataRequestUri, history: History | None = None) → HistoryDatasetAssociation

class galaxy.managers.hdas.HDAStorageCleanerManager(hda_manager: HDAManager, dataset_manager: DatasetManager)

Bases: StorageCleanerManager

__init__(hda_manager: HDAManager, dataset_manager: DatasetManager)

sort_map: Dict[StoredItemOrderBy, Any]

get_discarded_summary(user: User) → CleanableItemsSummary

Returns information with the total storage space taken by discarded items for the given user.

Discarded items are those that are deleted but not purged yet.

get_discarded(user: User, offset: int | None, limit: int | None, order: StoredItemOrderBy | None) → List[StoredItem]

Returns a paginated list of items deleted by the given user that are not yet purged.

cleanup_items(user: User, item_ids: Set[int]) → StorageItemsCleanupResult

Purges the given list of items by ID. The items must be owned by the user.

class galaxy.managers.hdas.HDASerializer(app: StructuredApp)

Bases: DatasetAssociationSerializer[HDAManager], TaggableSerializerMixin, AnnotatableSerializerMixin

model_manager_class

alias of HDAManager

app: StructuredApp

__init__(app: StructuredApp)

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

serialize_copied_from_ldda_id(item, key, **context)

Serialize an id attribute of item.

add_serializers()

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

serialize(item, keys, user=None, **context)

Override to hide information to users not able to access.

serialize_display_apps(item, key, trans=None, **context)

Return dictionary containing new-style display app urls.

serialize_old_display_applications(item, key, trans=None, **context)

Return dictionary containing old-style display app urls.

serialize_urls(item, key, **context)

Return web controller urls useful for this HDA.

class galaxy.managers.hdas.HDADeserializer(app: MinimalManagerApp)

Bases: DatasetAssociationDeserializer, TaggableDeserializerMixin, AnnotatableDeserializerMixin

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

model_manager_class

alias of HDAManager

__init__(app: MinimalManagerApp)

Set up deserializers and validator.

add_deserializers()

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

app: MinimalManagerApp

deserializable_keyset: Set[str]

class galaxy.managers.hdas.HDAFilterParser(app: MinimalManagerApp, **kwargs)

Bases: DatasetAssociationFilterParser, TaggableFilterMixin, AnnotatableFilterMixin

model_manager_class

alias of HDAManager

model_class

alias of HistoryDatasetAssociation

orm_filter_parsers: Dict[str, None | Dict[str, Any] | Callable]

fn_filter_parsers: Dict[str, Any]

dictionary containing parsing data for functional filters - applied after a query is made

app: MinimalManagerApp

galaxy.managers.hdcas module

Manager and Serializer for HDCAs.

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

galaxy.managers.hdcas.stream_dataset_collection(dataset_collection_instance, upstream_mod_zip=False, upstream_gzip=False)

galaxy.managers.hdcas.write_dataset_collection(dataset_collection_instance, archive)

galaxy.managers.hdcas.set_collection_attributes(dataset_element, *payload)

class galaxy.managers.hdcas.HDCAManager(app: MinimalManagerApp)

Bases: ModelManager, AccessibleManagerMixin, OwnableManagerMixin, PurgableManagerMixin, AnnotatableManagerMixin

Interface/service object for interacting with HDCAs.

model_class

alias of HistoryDatasetCollectionAssociation

foreign_key_name: str = 'history_dataset_collection_association'

tag_assoc

alias of HistoryDatasetCollectionTagAssociation

annotation_assoc

alias of HistoryDatasetCollectionAssociationAnnotationAssociation

__init__(app: MinimalManagerApp)

Set up and initialize other managers needed by hdas.

map_datasets(content, fn, *parents)

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

Uses the same kwargs as contents above.

update_attributes(content, payload: Dict)

app: BasicSharedApp

class galaxy.managers.hdcas.DCESerializer(app: StructuredApp)

Bases: ModelSerializer

Serializer for DatasetCollectionElements.

__init__(app: StructuredApp)

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

default_view: str | None

add_serializers()

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

serialize_object(item, key, **context)

views: Dict[str, List[str]]

serializable_keyset: Set[str]

serializers: Dict[str, Serializer]

model_manager_class: Type[T]

the class used to create this serializer’s generically accessible model_manager

app: MinimalManagerApp

class galaxy.managers.hdcas.DCSerializer(app: StructuredApp, dce_serializer=None)

Bases: ModelSerializer

Serializer for DatasetCollections.

__init__(app: StructuredApp, dce_serializer=None)

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

default_view: str | None

add_serializers()

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

serialize_elements(item, key, **context)

views: Dict[str, List[str]]

serializable_keyset: Set[str]

serializers: Dict[str, Serializer]

model_manager_class: Type[T]

the class used to create this serializer’s generically accessible model_manager

app: MinimalManagerApp

class galaxy.managers.hdcas.DCASerializer(app: StructuredApp, dce_serializer=None)

Bases: ModelSerializer

Base (abstract) Serializer class for HDCAs and LDCAs.

app: StructuredApp

__init__(app: StructuredApp, dce_serializer=None)

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

add_serializers()

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

class galaxy.managers.hdcas.HDCASerializer(app: StructuredApp)

Bases: DCASerializer, TaggableSerializerMixin, AnnotatableSerializerMixin

Serializer for HistoryDatasetCollectionAssociations.

__init__(app: StructuredApp)

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

default_view: str | None

add_serializers()

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

generate_contents_url(item, key, **context)

serialize_job_state_summary(item, key, **context)

serialize_elements_datatypes(item, key, **context)

app: StructuredApp

views: Dict[str, List[str]]

serializable_keyset: Set[str]

serializers: Dict[str, Serializer]

model_manager_class: Type[T]

the class used to create this serializer’s generically accessible model_manager

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.HistoryManager(app: MinimalManagerApp, hda_manager: HDAManager, contents_manager: HistoryContentsManager, contents_filters: HistoryContentsFilters)

Bases: SharableModelManager, PurgableManagerMixin, SortableManager

model_class

alias of History

foreign_key_name: str = 'history'

user_share_model

alias of HistoryUserShareAssociation

tag_assoc

alias of HistoryTagAssociation

annotation_assoc

alias of HistoryAnnotationAssociation

rating_assoc

alias of HistoryRatingAssociation

__init__(app: MinimalManagerApp, hda_manager: HDAManager, contents_manager: HistoryContentsManager, contents_filters: HistoryContentsFilters) → None

index_query(trans: ProvidesUserContext, payload: HistoryIndexQueryPayload, include_total_count: bool = False) → Tuple[List[History], int]

by_user(user: User, current_history: History | None = None, **kwargs: Any) → List[History]

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

is_owner(item: Base, user: User | None, current_history: History | None = None, **kwargs: Any) → bool

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

most_recent(user, filters=None, current_history=None)

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.

purge(item, flush=True, **kwargs)

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

get_current(trans)

Return the current history.

set_current(trans, history)

Set the current history.

set_current_by_id(trans, history_id)

Set the current history by an id.

parse_order_by(order_by_string, default=None)

Return an ORM compatible order_by using the given string

non_ready_jobs(history)

Return the currently running job objects associated with this history.

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

queue_history_import(trans, archive_type, archive_source)

legacy_serve_ready_history_export(trans, jeha)

get_ready_history_export_file_path(trans, jeha) → str

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

queue_history_export(trans, history, gzip=True, include_hidden=False, include_deleted=False, directory_uri=None, file_name=None)

get_sharing_extra_information(trans, item, users: Set[User], errors: Set[str], option: SharingOptions | None = None) → ShareHistoryExtra

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

is_history_shared_with(history: History, user: User) → bool

make_members_public(trans, item)

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

archive_history(history: History, archive_export_id: int | None)

Marks the history with the given id as archived and optionally associates it with the given archive export record.

Important: The caller is responsible for passing a valid archive_export_id that belongs to the given history.

restore_archived_history(history: History, force: bool = False)

Marks the history with the given id as not archived anymore.

Only un-archives the history if it is not associated with an archive export record. You can force the un-archiving in this case by passing force=True.

Please note that histories that are associated with an archive export are usually purged after export, so un-archiving them will not restore the datasets that were in the history before it was archived. You will need to import the archive export record to restore the history and its datasets as a new copy.

get_active_count(user: User) → int

Return the number of active histories for the given user.

class galaxy.managers.histories.HistoryStorageCleanerManager(history_manager: HistoryManager)

Bases: StorageCleanerManager

__init__(history_manager: HistoryManager)

sort_map: Dict[StoredItemOrderBy, Any]

get_discarded_summary(user: User) → CleanableItemsSummary

Returns information with the total storage space taken by discarded items for the given user.

Discarded items are those that are deleted but not purged yet.

get_discarded(user: User, offset: int | None, limit: int | None, order: StoredItemOrderBy | None) → List[StoredItem]

Returns a paginated list of items deleted by the given user that are not yet purged.

get_archived_summary(user: User) → CleanableItemsSummary

Returns information with the total storage space taken by archived items for the given user.

Archived items are those that are not currently active. Some archived items may be purged already, but this method does not return information about those.

get_archived(user: User, offset: int | None, limit: int | None, order: StoredItemOrderBy | None) → List[StoredItem]

Returns a paginated list of items archived by the given user that are not yet purged.

cleanup_items(user: User, item_ids: Set[int]) → StorageItemsCleanupResult

Purges the given list of items by ID. The items must be owned by the user.

class galaxy.managers.histories.HistoryExportManager(app: MinimalManagerApp, export_tracker: StoreExportTracker)

Bases: object

export_object_type = 'history'

__init__(app: MinimalManagerApp, export_tracker: StoreExportTracker)

get_task_exports(trans, history_id: int, limit: int | None = None, offset: int | None = None)

Returns task-based exports associated with this history

get_task_export_by_id(store_export_id: int) → StoreExportAssociation

create_export_association(history_id: int) → StoreExportAssociation

get_record_metadata(export: StoreExportAssociation) → ExportObjectMetadata | None

get_exports(trans, history_id: int)

Returns job-based exports associated with this history

serialize(trans, history_id: int, jeha: JobExportHistoryArchive) → dict

get_ready_jeha(trans, history_id: int, jeha_id: int | typing_extensions.Literal[latest] = 'latest')

class galaxy.managers.histories.HistorySerializer(app: MinimalManagerApp, hda_manager: HDAManager, hda_serializer: HDASerializer, history_contents_serializer: HistoryContentsSerializer)

Bases: SharableModelSerializer, PurgableSerializerMixin

Interface/service object for serializing histories into dictionaries.

model_manager_class

alias of HistoryManager

SINGLE_CHAR_ABBR: str | None = 'h'

__init__(app: MinimalManagerApp, hda_manager: HDAManager, hda_serializer: HDASerializer, history_contents_serializer: HistoryContentsSerializer)

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

add_serializers()

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

serialize_state_ids(item, key, **context)

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

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

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

serialize_history_state(item, key, **context)

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

serialize_contents(item, key, trans=None, user=None, **context)

serialize_contents_states(item, key, trans=None, **context)

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_contents_active(item, key, **context)

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.

class galaxy.managers.histories.HistoryDeserializer(app: MinimalManagerApp)

Bases: SharableModelDeserializer, PurgableDeserializerMixin

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

model_manager_class

alias of HistoryManager

__init__(app: MinimalManagerApp)

Set up deserializers and validator.

add_deserializers()

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

deserialize_preferred_object_store_id(item, key, val, **context)

app: MinimalManagerApp

deserializable_keyset: Set[str]

class galaxy.managers.histories.HistoryFilters(app: MinimalManagerApp, **kwargs)

Bases: SharableModelFilters, PurgableFiltersMixin

model_class

alias of History

model_manager_class

alias of HistoryManager

username_eq(item, val: str) → bool

username_contains(item, val: str) → bool

orm_filter_parsers: Dict[str, None | Dict[str, Any] | Callable]

fn_filter_parsers: Dict[str, Any]

dictionary containing parsing data for functional filters - applied after a query is made

app: MinimalManagerApp

galaxy.managers.histories.get_count(session, statement)

galaxy.managers.history_contents module

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

class galaxy.managers.history_contents.HistoryContentsManager(app: MinimalManagerApp)

Bases: SortableManager

root_container_class

alias of History

contained_class

alias of HistoryDatasetAssociation

contained_class_manager_class

alias of HDAManager

contained_class_type_name = 'dataset'

subcontainer_class

alias of HistoryDatasetCollectionAssociation

subcontainer_class_manager_class

alias of HDCAManager

subcontainer_class_type_name = 'dataset_collection'

common_columns = ('history_id', 'history_content_type', 'id', 'type_id', 'hid', 'extension', 'dataset_id', 'collection_id', 'name', 'state', 'object_store_id', 'size', 'deleted', 'purged', 'visible', 'create_time', 'update_time')

default_order_by = 'hid'

__init__(app: MinimalManagerApp)

contents(container, filters=None, limit=None, offset=None, order_by=None, **kwargs)

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)

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)

Returns the contents union query for subqueries, etc.

parse_order_by(order_by_string, default=None)

Return an ORM compatible order_by using the given string

state_counts(history)

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

Note: does not include deleted/hidden contents.

active_counts(history)

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.

map_datasets(history, fn, **kwargs)

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

Uses the same kwargs as contents above.

static passes_filters(content, filters)

class galaxy.managers.history_contents.HistoryContentsSerializer(app: MinimalManagerApp, **kwargs)

Bases: ModelSerializer, PurgableSerializerMixin

Interface/service object for serializing histories into dictionaries.

model_manager_class

alias of HistoryContentsManager

__init__(app: MinimalManagerApp, **kwargs)

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

default_view: str | None

add_serializers()

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

serialize_id_or_skip(item: Any, key: str, **context)

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

views: Dict[str, List[str]]

serializable_keyset: Set[str]

serializers: Dict[str, Serializer]

app: MinimalManagerApp

class galaxy.managers.history_contents.HistoryContentsFilters(app: MinimalManagerApp, **kwargs)

Bases: ModelFilterParser, AnnotatableFilterMixin, PurgableFiltersMixin, GenomeFilterMixin, TaggableFilterMixin, ToolFilterMixin

model_class

alias of HistoryDatasetAssociation

parse_query_filters_with_relations(query_filters: ValueFilterQueryParams, history_id)

Parse query filters but consider case where related filter is included.

get_query_filters_with_relations(query_filters: ValueFilterQueryParams, related_q: str, history_id)

Return query_filters_with_relations changing related:hid to related:[hid1, hid2, …].

decode_type_id(type_id)

parse_type_id_list(type_id_list_string, sep=',')

Split type_id_list_string at sep.

orm_filter_parsers: Dict[str, None | Dict[str, Any] | Callable]

fn_filter_parsers: Dict[str, Any]

dictionary containing parsing data for functional filters - applied after a query is made

model_manager_class: Type[T]

the class used to create this serializer’s generically accessible model_manager

app: MinimalManagerApp

galaxy.managers.interactivetool module

class galaxy.managers.interactivetool.InteractiveToolPropagatorSQLAlchemy(database_url, encode_id)

Bases: object

Propagator for InteractiveToolManager implemented using SQLAlchemy.

__init__(database_url, encode_id)

Constructor that sets up the propagator using a SQLAlchemy database URL.

Parameters:

• database_url – SQLAlchemy database URL, read more on the SQLAlchemy documentation https://docs.sqlalchemy.org/en/20/core/engines.html#database-urls.

• encode_id – A helper class that can encode ids as lowercase alphanumeric strings and vice versa.

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

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

remove(**kwd)

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

save_entry_point(entry_point)

Convenience method to easily save an entry_point.

remove_entry_point(entry_point)

Convenience method to easily remove an entry_point.

class galaxy.managers.interactivetool.InteractiveToolManager(app)

Bases: object

Manager for dealing with InteractiveTools

__init__(app)

create_entry_points(job, tool, entry_points=None, flush=True)

configure_entry_point(job, tool_port=None, host=None, port=None, protocol=None)

configure_entry_points(job, ports_dict)

save_entry_point(entry_point)

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

create_interactivetool(job, tool, entry_points)

get_nonterminal_for_user_by_trans(trans)

can_access_job(trans, job)

can_access_entry_point(trans, entry_point)

can_access_entry_points(trans, entry_points)

stop(trans, entry_point)

remove_entry_points(entry_points)

remove_entry_point(entry_point, flush=True)

target_if_active(trans, entry_point)

get_entry_point_subdomain(trans, entry_point)

get_entry_point_path(trans, entry_point)

access_entry_point_target(trans, entry_point_id)

galaxy.managers.jobs module

class galaxy.managers.jobs.JobLock(*, active: bool)

Bases: BaseModel

active: bool

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'active': FieldInfo(annotation=bool, required=True, title='Job lock status', description='If active, jobs will not dispatch')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

galaxy.managers.jobs.get_path_key(path_tuple)

class galaxy.managers.jobs.JobManager(app: StructuredApp)

Bases: object

__init__(app: StructuredApp)

index_query(trans: ProvidesUserContext, payload: JobIndexQueryPayload) → ScalarResult

The caller is responsible for security checks on the resulting job if history_id, invocation_id, or implicit_collection_jobs_id is set. Otherwise this will only return the user’s jobs or all jobs if the requesting user is acting as an admin.

job_lock() → JobLock

update_job_lock(job_lock: JobLock)

get_accessible_job(trans: ProvidesUserContext, decoded_job_id) → Job

get_job_console_output(trans, job, stdout_position=-1, stdout_length=0, stderr_position=-1, stderr_length=0)

stop(job, message=None)

class galaxy.managers.jobs.JobSearch(sa_session: galaxy_scoped_session, hda_manager: HDAManager, dataset_collection_manager: DatasetCollectionManager, ldda_manager: LDDAManager, id_encoding_helper: IdEncodingHelper)

Bases: object

Search for jobs using tool inputs or other jobs

__init__(sa_session: galaxy_scoped_session, hda_manager: HDAManager, dataset_collection_manager: DatasetCollectionManager, ldda_manager: LDDAManager, id_encoding_helper: IdEncodingHelper)

by_tool_input(trans: ProvidesHistoryContext, tool_id: str, tool_version: str | None, param: Dict[str, Any], param_dump: Dict[str, Any], job_state: str | List[str] | None = 'ok')

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

galaxy.managers.jobs.view_show_job(trans, job: Job, full: bool) → Dict

galaxy.managers.jobs.invocation_job_source_iter(sa_session, invocation_id)

galaxy.managers.jobs.get_job_metrics_for_invocation(sa_session: galaxy_scoped_session, invocation_id: int)

galaxy.managers.jobs.fetch_job_states(sa_session, job_source_ids, job_source_types)

galaxy.managers.jobs.summarize_invocation_jobs(invocation_id, job_summaries, implicit_collection_job_summaries, invocation_state, invocation_step_states)

class galaxy.managers.jobs.JobsSummary

Bases: TypedDict

populated_state: str

states: Dict[str, int]

model: str

id: int

galaxy.managers.jobs.summarize_jobs_to_dict(sa_session, jobs_source) → JobsSummary | None

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.summarize_job_metrics(trans, job)

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_metrics(trans: ProvidesUserContext, job_metrics)

galaxy.managers.jobs.summarize_destination_params(trans, job)

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_job_parameters(trans, job: Job)

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.get_output_name(tool, output, params)

galaxy.managers.jobs.summarize_job_outputs(job: Job, tool, params)

galaxy.managers.jobs.get_jobs_to_check_at_startup(session: galaxy_scoped_session, track_jobs_in_database: bool, config)

galaxy.managers.jobs.get_job(session, *where_clauses)

galaxy.managers.lddas module

class galaxy.managers.lddas.LDDAManager(app: MinimalManagerApp)

Bases: DatasetAssociationManager[LibraryDatasetDatasetAssociation]

A fairly sparse manager for LDDAs.

model_class

alias of LibraryDatasetDatasetAssociation

__init__(app: MinimalManagerApp)

Set up and initialize other managers needed by lddas.

get(trans, id: int, check_accessible=True) → LibraryDatasetDatasetAssociation

app: MinimalManagerApp

foreign_key_name: str

galaxy.managers.libraries module

Manager and Serializer for libraries.

class galaxy.managers.libraries.LibraryManager

Bases: object

Interface/service object for interacting with libraries.

get(trans, decoded_library_id: int, check_accessible: bool = True) → Library

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

create(trans, name: str, description: str | None = '', synopsis: str | None = '') → Library

Create a new library.

update(trans, library: Library, name: str | None = None, description: str | None = None, synopsis: str | None = None) → Library

Update the given library

delete(trans, library: Library, undelete: bool | None = False) → Library

Mark given library deleted/undeleted based on the flag.

list(trans, deleted: bool | None = False) → Tuple[Query, Dict[str, Set]]

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:

iterable that will emit all accessible libraries

Return type:

sqlalchemy ScalarResult

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

secure(trans, library: Library, check_accessible: bool = True) → Library

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

check_accessible(trans, library: Library) → Library

Check whether the library is accessible to current user.

get_library_dict(trans, library: Library, prefetched_ids: Dict[str, Set] | None = None) → dict

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_current_roles(trans, library: Library) → dict

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_access_roles(trans, library: Library) → Set[Role]

Load access roles for all library permissions

get_modify_roles(trans, library: Library) → Set[Role]

Load modify roles for all library permissions

get_manage_roles(trans, library: Library) → Set[Role]

Load manage roles for all library permissions

get_add_roles(trans, library: Library) → Set[Role]

Load add roles for all library permissions

make_public(trans, library: Library) → bool

Makes the given library public (removes all access roles)

is_public(trans, library: Library) → bool

Return true if lib is public.

galaxy.managers.libraries.get_containing_library_from_library_dataset(trans, library_dataset) → Library | None

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: MinimalManagerApp)

Bases: ModelManager[LibraryDataset]

Interface/service object for interacting with library datasets.

model_class

alias of LibraryDataset

__init__(app: MinimalManagerApp)

get(trans, decoded_library_dataset_id, check_accessible=True) → LibraryDataset

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

update(item: LibraryDataset, new_values: Dict[str, Any], flush: bool = True, **kwargs) → LibraryDataset

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

Parameters:

• item (LibraryDataset) – library dataset to change

• new_values (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

secure(trans, ld, check_accessible=True, check_ownership=False)

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

Parameters:

• ld (galaxy.model.LibraryDataset) – library dataset

• check_accessible (bool) – flag whether to check that user can access library dataset

Returns:

the original library dataset

Return type:

galaxy.model.LibraryDataset

check_accessible(trans, ld)

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)

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

serialize(trans, ld: LibraryDataset) → Dict[str, Any]

Serialize the library dataset into a dictionary.

foreign_key_name: str

app: BasicSharedApp

galaxy.managers.library_datasets.get_library_dataset(session, library_dataset_id) → LibraryDataset

galaxy.managers.licenses module

class galaxy.managers.licenses.LicenseMetadataModel(*, licenseId: str, name: str, reference: str, referenceNumber: int, isDeprecatedLicenseId: bool, isOsiApproved: bool, seeAlso: List[Url[Url]], detailsUrl: Url[Url], recommended: bool, url: Url[Url], spdxUrl: Url[Url])

Bases: BaseModel

licenseId: str

name: str

reference: str

referenceNumber: int

isDeprecatedLicenseId: bool

isOsiApproved: bool

seeAlso: List[Url[Url]]

detailsUrl: Url[Url]

recommended: bool

url: Url[Url]

spdxUrl: Url[Url]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'detailsUrl': FieldInfo(annotation=Url, required=True, title='Details URL', description='URL to the SPDX json details for this license', examples=['http://spdx.org/licenses/Apache-2.0.json'], metadata=[UrlConstraints(max_length=2083, allowed_schemes=['http', 'https'], host_required=None, default_host=None, default_port=None, default_path=None)]), 'isDeprecatedLicenseId': FieldInfo(annotation=bool, required=True, title='Deprecated License', description='True if the entire license is deprecated', examples=[False]), 'isOsiApproved': FieldInfo(annotation=bool, required=True, title='OSI approved', description='Indicates if the [OSI](https://opensource.org/) has approved the license', examples=[True]), 'licenseId': FieldInfo(annotation=str, required=True, title='Identifier', description='SPDX Identifier', examples=['Apache-2.0']), 'name': FieldInfo(annotation=str, required=True, title='Name', description='Full name of the license', examples=['Apache License 2.0']), 'recommended': FieldInfo(annotation=bool, required=True, title='Recommended', description='True if this license is recommended to be used'), 'reference': FieldInfo(annotation=str, required=True, title='Reference', description='Reference to the HTML format for the license file', examples=['./Apache-2.0.html']), 'referenceNumber': FieldInfo(annotation=int, required=True, title='Reference number', description='*Deprecated* - this field is generated and is no longer in use'), 'seeAlso': FieldInfo(annotation=List[Annotated[pydantic_core._pydantic_core.Url, UrlConstraints(max_length=2083, allowed_schemes=['http', 'https'], host_required=None, default_host=None, default_port=None, default_path=None)]], required=True, title='Reference URLs', description='Cross reference URL pointing to additional copies of the license'), 'spdxUrl': FieldInfo(annotation=Url, required=True, title='SPDX URL', examples=['https://spdx.org/licenses/Apache-2.0.html'], metadata=[UrlConstraints(max_length=2083, allowed_schemes=['http', 'https'], host_required=None, default_host=None, default_port=None, default_path=None)]), 'url': FieldInfo(annotation=Url, required=True, title='URL', description='License URL', examples=['http://www.apache.org/licenses/LICENSE-2.0'], metadata=[UrlConstraints(max_length=2083, allowed_schemes=['http', 'https'], host_required=None, default_host=None, default_port=None, default_path=None)])}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class galaxy.managers.licenses.LicensesManager

Bases: object

__init__()

index()

get(uri)

get_licenses() → List[LicenseMetadataModel]

get_license_by_id(id: str) → LicenseMetadataModel

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)

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: PdfDocumentType) → bytes

galaxy.managers.markdown_util.ready_galaxy_markdown_for_export(trans, internal_galaxy_markdown)

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)

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

galaxy.managers.markdown_util.resolve_invocation_markdown(trans, invocation, workflow_markdown)

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.markdown_util.to_basic_markdown(trans, internal_galaxy_markdown: str) → str

Replace Galaxy Markdown extensions with plain Markdown for PDF/HTML export.

galaxy.managers.metrics module

class galaxy.managers.metrics.Metric(*, namespace: str, time: str, level: int, args: str)

Bases: BaseModel

namespace: str

time: str

level: int

args: str

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'args': FieldInfo(annotation=str, required=True, title='Arguments', description='A JSON string containing an array of extra data.'), 'level': FieldInfo(annotation=int, required=True, title='Level', description="An integer representing the metric's log level."), 'namespace': FieldInfo(annotation=str, required=True, title='Namespace', description='Label indicating the source of the metric.'), 'time': FieldInfo(annotation=str, required=True, title='Timestamp', description='The timestamp in ISO format.', examples=['2021-01-23T18:25:43.511Z'])}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class galaxy.managers.metrics.CreateMetricsPayload(*, metrics: List[Metric] = [])

Bases: BaseModel

metrics: List[Metric]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'metrics': FieldInfo(annotation=List[galaxy.managers.metrics.Metric], required=False, default=[], title='List of metrics to be recorded.', examples=[{'namespace': 'test-source', 'time': '2021-01-23T18:25:43.511Z', 'level': 0, 'args': '{"test":"value"}'}])}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class galaxy.managers.metrics.MetricsManager(app: MinimalManagerApp)

Bases: object

Interface/service object shared by controllers for interacting with metrics.

__init__(app: MinimalManagerApp) → None

debugging

set to true to send additional debugging info to the log

create(trans, payload: CreateMetricsPayload)

Record any metrics sent and return some status object.

Note

Anonymous users can post metrics

Parameters:

payload (dict) –

(optional) dictionary structure containing: * metrics: a list containing dictionaries of the form

namespace: label indicating the source of the metric time: isoformat datetime when the metric was recorded level: an integer representing the metric’s log level args: a json string containing an array of extra data

Return type:

dict

Returns:

status object

galaxy.managers.model_stores module

class galaxy.managers.model_stores.ModelStoreUserContext(app: MinimalManagerApp, user: User)

Bases: ProvidesUserContext

__init__(app: MinimalManagerApp, user: User) → None

property app

Provide access to the Galaxy app object.

property 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).

get_user()

set_user(user)

property user

Provide access to the user object.

class galaxy.managers.model_stores.ModelStoreManager(app: MinimalManagerApp, history_manager: HistoryManager, export_tracker: StoreExportTracker, sa_session: galaxy_scoped_session, job_manager: JobManager, short_term_storage_monitor: ShortTermStorageMonitor, user_manager: UserManager)

Bases: object

__init__(app: MinimalManagerApp, history_manager: HistoryManager, export_tracker: StoreExportTracker, sa_session: galaxy_scoped_session, job_manager: JobManager, short_term_storage_monitor: ShortTermStorageMonitor, user_manager: UserManager)

setup_history_export_job(request: SetupHistoryExportJob)

prepare_history_download(request: GenerateHistoryDownload)

prepare_history_content_download(request: GenerateHistoryContentDownload)

prepare_invocation_download(request: GenerateInvocationDownload)

write_invocation_to(request: WriteInvocationTo)

write_history_content_to(request: WriteHistoryContentTo)

write_history_to(request: WriteHistoryTo)

set_history_export_request_metadata(request: WriteHistoryTo | GenerateHistoryDownload) → ExportObjectMetadata | None

set_history_export_result_metadata(export_association_id: int | None, export_metadata: ExportObjectMetadata | None, success: bool, error: str | None = None)

import_model_store(request: ImportModelStoreTaskRequest)

galaxy.managers.model_stores.create_objects_from_store(app: MinimalManagerApp, galaxy_user: User | None, payload, history: History | None = None, for_library: bool = False) → ObjectImportTracker

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.PageManager(app: MinimalManagerApp)

Bases: SharableModelManager, UsesAnnotations

Provides operations for managing a Page.

model_class

alias of Page

foreign_key_name: str = 'page'

user_share_model

alias of PageUserShareAssociation

tag_assoc

alias of PageTagAssociation

annotation_assoc

alias of PageAnnotationAssociation

rating_assoc

alias of PageRatingAssociation

__init__(app: MinimalManagerApp)

index_query(trans: ProvidesUserContext, payload: PageIndexQueryPayload, include_total_count: bool = False) → Tuple[Result, int]

create_page(trans, payload: CreatePagePayload)

save_new_revision(trans, page, payload)

rewrite_content_for_import(trans, content, content_format: str)

rewrite_content_for_export(trans, as_dict)

class galaxy.managers.pages.PageSerializer(app: MinimalManagerApp)

Bases: SharableModelSerializer

Interface/service object for serializing pages into dictionaries.

model_manager_class

alias of PageManager

SINGLE_CHAR_ABBR: str | None = 'p'

__init__(app: MinimalManagerApp)

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

add_serializers()

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

class galaxy.managers.pages.PageDeserializer(app: MinimalManagerApp)

Bases: SharableModelDeserializer

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

model_manager_class

alias of PageManager

__init__(app: MinimalManagerApp)

Set up deserializers and validator.

add_deserializers()

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

app: MinimalManagerApp

deserializers: Dict[str, Deserializer]

deserializable_keyset: Set[str]

class galaxy.managers.pages.PageContentProcessor(trans, render_embed_html_fn: Callable)

Bases: HTMLParser

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

bare_ampersand = re.compile('&(?!#\\d+;|#x[0-9a-fA-F]+;|\\w+;)')

elements_no_end_tag = {'area', 'base', 'basefont', 'br', 'col', 'command', 'embed', 'frame', 'hr', 'img', 'input', 'isindex', 'keygen', 'link', 'meta', 'param', 'source', 'track', 'wbr'}

__init__(trans, render_embed_html_fn: Callable)

Initialize and reset this instance.

If convert_charrefs is True (the default), all character references are automatically converted to the corresponding Unicode characters.

reset()

Reset this instance. Loses all unprocessed data.

feed(data)

Feed data to the parser.

Call this as often as you want, with as little or as much text as you want (may include ‘n’).

handle_starttag(tag, attrs)

Called for each start tag

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

handle_endtag(tag)

Called for each end tag

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

handle_charref(ref)

handle_entityref(ref)

handle_data(text)

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_comment(text)

handle_decl(text)

handle_pi(text)

output()

Return processed HTML as a single string

galaxy.managers.pages.get_page_identifiers(item_id, app)

galaxy.managers.pages.placeholderRenderForEdit(trans: ProvidesHistoryContext, item_class, item_id)

galaxy.managers.pages.placeholderRenderForSave(trans: ProvidesHistoryContext, item_class, item_id, encode=False)

galaxy.managers.pages.get_page_revision(session: galaxy_scoped_session, page_id: int)

galaxy.managers.pages.get_shared_pages(session: galaxy_scoped_session, user: User)

galaxy.managers.pages.get_page(session: galaxy_scoped_session, user: User, slug: str)

galaxy.managers.pages.page_exists(session: galaxy_scoped_session, user: User, slug: str) → bool

galaxy.managers.pages.get_count(session, statement)

galaxy.managers.quotas module

Manager and Serializers for Quotas.

For more information about quotas: https://galaxyproject.org/admin/disk-quotas/

class galaxy.managers.quotas.QuotaManager(app: StructuredApp)

Bases: object

Interface/service object to interact with Quotas.

__init__(app: StructuredApp)

property sa_session

property quota_agent: DatabaseQuotaAgent

create_quota(payload: dict, decode_id=None) → Tuple[Quota, str]

rename_quota(quota, params) → str | None

manage_users_and_groups_for_quota(quota, params, decode_id=None) → str | None

edit_quota(quota, params) → str | None

set_quota_default(quota, params) → str | None

unset_quota_default(quota, params=None) → str | None

delete_quota(quota, params=None) → str

undelete_quota(quota, params=None) → str

purge_quota(quota, params=None) → str

This method should only be called for a Quota that has previously been deleted. Purging a deleted Quota deletes all of the following from the database: - UserQuotaAssociations where quota_id == Quota.id - GroupQuotaAssociations where quota_id == Quota.id

get_quota(trans, id: int, deleted: bool | None = None) → Quota

galaxy.managers.ratable module

Mixins for Ratable model managers and serializers.

class galaxy.managers.ratable.RatableManagerMixin

Bases: object

rating_assoc: Type[ItemRatingAssociation]

rating(item, user, as_int=True)

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

Returns the full rating model if as_int is False.

ratings(item)

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

ratings_avg(item)

Returns the average of all ratings given to this item.

ratings_count(item)

Returns the number of ratings given to this item.

rate(item, user, value, flush=True)

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

class galaxy.managers.ratable.RatableSerializerMixin

Bases: object

add_serializers()

serialize_user_rating(item, key, user=None, **context)

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

serialize_community_rating(item, key, **context)

Returns a dictionary containing:

average the (float) average of all ratings of this object count the number of ratings

class galaxy.managers.ratable.RatableDeserializerMixin

Bases: object

add_deserializers()

deserialize_rating(item, key, val, user=None, **context)

class galaxy.managers.ratable.RatableFilterMixin

Bases: object

galaxy.managers.rbac_secured module

exception galaxy.managers.rbac_secured.RBACPermissionFailedException(err_msg: str | None = None, type='info', **extra_error_info)

Bases: InsufficientPermissionsException

class galaxy.managers.rbac_secured.RBACPermission(app)

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).

permissions_class: type

permission_failed_error_class

alias of RBACPermissionFailedException

__init__(app)

session()

is_permitted(item, user, trans=None)

error_unless_permitted(item, user, trans=None)

grant(item, user, flush: bool = True)

revoke(item, user, flush: bool = True)

class galaxy.managers.rbac_secured.DatasetRBACPermission(app)

Bases: 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

permissions_class

alias of DatasetPermissions

action_name = None

by_dataset(dataset)

by_roles(dataset, roles)

by_role(dataset, role)

set(dataset, roles, flush=True)

clear(dataset, flush=True)

galaxy.managers.rbac_secured.iterable_has_all(iterable, has_these)

exception galaxy.managers.rbac_secured.DatasetManagePermissionFailedException(err_msg: str | None = None, type='info', **extra_error_info)

Bases: RBACPermissionFailedException

class galaxy.managers.rbac_secured.ManageDatasetRBACPermission(app)

Bases: 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'

permission_failed_error_class

alias of DatasetManagePermissionFailedException

is_permitted(dataset, user, trans=None)

grant(item, user, flush: bool = True)

revoke(item, user, flush: bool = True)

exception galaxy.managers.rbac_secured.DatasetAccessPermissionFailedException(err_msg: str | None = None, type='info', **extra_error_info)

Bases: RBACPermissionFailedException

class galaxy.managers.rbac_secured.AccessDatasetRBACPermission(app)

Bases: 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'

permission_failed_error_class

alias of DatasetAccessPermissionFailedException

is_permitted(dataset, user, trans=None)

grant(item, user, flush: bool = True)

revoke(item, user, flush: bool = True)

is_public(dataset)

set_private(dataset, user, flush=True)

galaxy.managers.remote_files module

class galaxy.managers.remote_files.RemoteFilesManager(app: MinimalManagerApp)

Bases: object

Interface/service object for interacting with remote files.

__init__(app: MinimalManagerApp)

index(user_ctx: ProvidesUserContext, target: str, format: RemoteFilesFormat | None, recursive: bool | None, disable: RemoteFilesDisableMode | None, writeable: bool | None = False, limit: int | None = None, offset: int | None = None, query: str | None = None, sort_by: str | None = None) → Tuple[ListUriResponse | ListJstreeResponse, int]

Returns a list of remote files and directories available to the user and the total count of them.

get_files_source_plugins(user_context: ProvidesUserContext, browsable_only: bool | None = True, include_kind: Set[PluginKind] | None = None, exclude_kind: Set[PluginKind] | None = None)

Display plugin information for each of the gxfiles:// URI targets available.

create_entry(user_ctx: ProvidesUserContext, entry_data: CreateEntryPayload) → CreatedEntryResponse

Create an entry (directory or record) in a remote files location.

galaxy.managers.roles module

Manager and Serializer for Roles.

class galaxy.managers.roles.RoleManager(app: BasicSharedApp)

Bases: ModelManager[Role]

Business logic for roles.

model_class

alias of Role

foreign_key_name: str = 'role'

user_assoc

alias of UserRoleAssociation

group_assoc

alias of GroupRoleAssociation

get(trans: ProvidesUserContext, role_id: int) → Role

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

Parameters:

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

list_displayable_roles(trans: ProvidesUserContext) → List[Role]

create_role(trans: ProvidesUserContext, role_definition_model: RoleDefinitionModel) → Role

delete(trans: ProvidesUserContext, role: Role) → Role

purge(trans: ProvidesUserContext, role: Role) → Role

undelete(trans: ProvidesUserContext, role: Role) → Role

app: BasicSharedApp

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

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.

model_class: Type[Any]

by_id(id: int)

is_accessible(item, user: User, **kwargs: Any) → bool

Return True if the item accessible to user.

get_accessible(id: int, user: User, **kwargs: Any)

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

Raises:

exceptions.ItemAccessibilityException –

error_unless_accessible(item: Query, user, **kwargs)

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

Raises:

exceptions.ItemAccessibilityException –

list_accessible(user, **kwargs)

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

Raises:

exceptions.ItemAccessibilityException –

filter_accessible(user, **kwargs)

Return a list of items accessible to the user.

class galaxy.managers.secured.OwnableManagerMixin

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.

model_class: Type[Any]

by_id(id: int)

is_owner(item: Base, user: User | None, **kwargs: Any) → bool

Return True if user owns the item.

get_owned(id: int, user: User | None, **kwargs: Any) → Any

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

Raises:

exceptions.ItemOwnershipException –

error_unless_owner(item, user: User | None, **kwargs: Any)

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

Raises:

exceptions.ItemAccessibilityException –

list_owned(user, **kwargs)

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

Raises:

exceptions.ItemAccessibilityException –

filter_owned(user, **kwargs)

Return a list of items owned by the user.

get_mutable(id: int, user: User | None, **kwargs: Any) → Any

Return the item with the given id if the user can mutate it, otherwise raise an error. The user must be the owner of the item.

Raises:

exceptions.ItemOwnershipException –

error_unless_mutable(item)

Raise an error if the item is NOT mutable.

Items purged or archived are considered immutable.

Raises:

exceptions.ItemImmutableException –

galaxy.managers.session module

class galaxy.managers.session.GalaxySessionManager(model: SharedModelMapping)

Bases: object

Manages GalaxySession.

__init__(model: SharedModelMapping)

get_session_from_session_key(session_key: str)

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: MinimalManagerApp, **kwargs)

Bases: ModelDeserializer, TaggableDeserializerMixin, AnnotatableDeserializerMixin, RatableDeserializerMixin

__init__(app: MinimalManagerApp, **kwargs)

Set up deserializers and validator.

add_deserializers()

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

deserialize_published(item, key, val, **context)

deserialize_importable(item, key, val, **context)

deserialize_users_shared_with(item, key, val, **context)

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.

app: MinimalManagerApp

deserializers: Dict[str, Deserializer]

deserializable_keyset: Set[str]

model_manager_class: Type[T]

the class used to create this serializer’s generically accessible model_manager

class galaxy.managers.sharable.SharableModelFilters(app: MinimalManagerApp, **kwargs)

Bases: ModelFilterParser, TaggableFilterMixin, AnnotatableFilterMixin, RatableFilterMixin

model_class: Type[object]

orm_filter_parsers: Dict[str, None | Dict[str, Any] | Callable]

fn_filter_parsers: Dict[str, Any]

dictionary containing parsing data for functional filters - applied after a query is made

model_manager_class: Type[T]

the class used to create this serializer’s generically accessible model_manager

app: MinimalManagerApp

class galaxy.managers.sharable.SharableModelManager(app: MinimalManagerApp)

Bases: ModelManager, OwnableManagerMixin, AccessibleManagerMixin, AnnotatableManagerMixin, RatableManagerMixin

user_share_model: Type[UserShareAssociation]

the model used for UserShareAssociations with this model

SINGLE_CHAR_ABBR: str | None = None

the single character abbreviation used in username_and_slug: e.g. ‘h’ for histories: u/user/h/slug

__init__(app: MinimalManagerApp)

by_user(user: User, **kwargs: Any) → List[Any]

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

is_owner(item: Base, user: User | None, **kwargs: Any) → bool

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

is_accessible(item, user: User | None, **kwargs: Any) → bool

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

make_importable(item, flush=True)

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_non_importable(item, flush=True)

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)

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

unpublish(item, flush=True)

Set the published flag on item to False.

list_published(filters=None, **kwargs)

Return a list of all published items.

get_share_assocs(item, user=None)

Get the UserShareAssociations for the item.

Optionally send in user to test for a single match.

share_with(item, user: User, flush: bool = True)

Get or create a share for the given user.

unshare_with(item, user: User, flush: bool = True)

Delete a user share from the database.

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

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

get_sharing_extra_information(trans, item, users: Set[User], errors: Set[str], option: SharingOptions | None = None) → ShareWithExtra | None

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.

make_members_public(trans, item)

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.

update_current_sharing_with_users(item, new_users_shared_with: Set[User], flush=True)

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

set_slug(item, new_slug, user, flush=True)

Validate and set the new slug for item.

get_unique_slug(item)

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

create_unique_slug(item, flush=True)

Set a new, unique slug on the item.

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

Bases: ModelSerializer, TaggableSerializerMixin, AnnotatableSerializerMixin, RatableSerializerMixin

SINGLE_CHAR_ABBR: str | None = None

__init__(app, **kwargs)

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

add_serializers()

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

serialize_email_hash(item, key, **context)

serialize_title(item, key, **context)

serialize_username(item, key, **context)

serialize_username_and_slug(item, key, **context)

serialize_users_shared_with(item, key, user=None, **context)

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.SharingOptions(value)

Bases: str, Enum

Options for sharing resources that may have restricted access to all or part of their contents.

make_public = 'make_public'

make_accessible_to_shared = 'make_accessible_to_shared'

no_changes = 'no_changes'

class galaxy.managers.sharable.ShareWithExtra(*, can_share: bool = False)

Bases: Model

can_share: bool

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'protected_namespaces': (), 'use_enum_values': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'can_share': FieldInfo(annotation=bool, required=False, default=False, title='Can Share', description='Indicates whether the resource can be directly shared or requires further actions.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class galaxy.managers.sharable.SlugBuilder

Bases: object

Builder for creating slugs out of items.

create_item_slug(sa_session, item) → bool

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

classmethod is_valid_slug(slug)

Returns true if slug is valid.

galaxy.managers.taggable module

Mixins for Taggable model managers and serializers.

class galaxy.managers.taggable.TaggableSerializerMixin

Bases: object

add_serializers()

serialize_tags(item, key, **context)

Return tags as a list of strings.

class galaxy.managers.taggable.TaggableDeserializerMixin

Bases: object

tag_handler: GalaxyTagHandler

validate: ModelValidator

add_deserializers()

deserialize_tags(item, key, val, *, user: User | None = None, trans: ProvidesUserContext, **context)

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

Bases: object

valid_ops = ('eq', 'contains', 'has')

create_tag_filter(attr, op, val)

galaxy.managers.tags module

class galaxy.managers.tags.TaggableItemClass(value)

Bases: Enum

An enumeration.

History = 'History'

HistoryDatasetAssociation = 'HistoryDatasetAssociation'

HistoryDatasetCollectionAssociation = 'HistoryDatasetCollectionAssociation'

LibraryDatasetDatasetAssociation = 'LibraryDatasetDatasetAssociation'

Page = 'Page'

StoredWorkflow = 'StoredWorkflow'

Visualization = 'Visualization'

class galaxy.managers.tags.ItemTagsPayload(*, item_id: int[int], item_class: TaggableItemClass, item_tags: TagCollection | None = None)

Bases: Model

item_id: int[int]

item_class: TaggableItemClass

item_tags: TagCollection | None

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'protected_namespaces': (), 'use_enum_values': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'item_class': FieldInfo(annotation=TaggableItemClass, required=True, title='Item class', description='The name of the class of the item that will be tagged.'), 'item_id': FieldInfo(annotation=int, required=True, title='Item ID', description='The `encoded identifier` of the item whose tags will be updated.', metadata=[BeforeValidator(func=<function <lambda>>, json_schema_input_type=PydanticUndefined), BeforeValidator(func=<function ensure_valid_id>, json_schema_input_type=PydanticUndefined), PlainSerializer(func=<function <lambda>>, return_type=<class 'str'>, when_used='json'), WithJsonSchema(json_schema={'type': 'string', 'example': '0123456789ABCDEF', 'pattern': '[0-9a-fA-F]+', 'minLength': 16}, mode='serialization'), WithJsonSchema(json_schema={'type': 'string', 'example': '0123456789ABCDEF', 'pattern': '[0-9a-fA-F]+', 'minLength': 16}, mode='validation')]), 'item_tags': FieldInfo(annotation=Union[TagCollection, NoneType], required=False, default=None, title='Item tags', description='The list of tags that will replace the current tags associated with the item.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class galaxy.managers.tags.TagsManager

Bases: object

Interface/service object shared by controllers for interacting with tags.

update(trans: ProvidesUserContext, payload: ItemTagsPayload) → None

Apply a new set of tags to an item; previous tags are deleted.

galaxy.managers.tool_data module

class galaxy.managers.tool_data.ToolDataManager(app: StructuredApp)

Bases: object

Interface/service object for interacting with tool data.

__init__(app: StructuredApp)

property data_tables: Dict[str, ToolDataTable]

index() → ToolDataEntryList

Return all tool data tables.

show(table_name: str) → ToolDataDetails

Get details of a given data table

show_field(table_name: str, field_name: str) → ToolDataField

Get information about a partiular field in a tool data table

reload(table_name: str) → ToolDataDetails

Reloads a tool data table.

get_field_file_path(table_name: str, field_name: str, file_name: str) → Path

Get the absolute path to a given file name in the table field

delete(table_name: str, values: str | None = None) → ToolDataDetails

Removes an item from a data table

class galaxy.managers.tool_data.ToolDataImportManager(app: MinimalManagerApp)

Bases: object

__init__(app: MinimalManagerApp)

file_sources: ConfiguredFileSources

tool_data_tables: ToolDataTableManager

import_data_bundle_by_uri(config, uri: str, tool_data_file_path=None)

import_data_bundle_by_dataset(config, dataset: DatasetInstance, tool_data_file_path=None)

galaxy.managers.tools module

class galaxy.managers.tools.DynamicToolManager(app: BasicSharedApp)

Bases: ModelManager

Manages dynamic tools stored in Galaxy’s database.

model_class

alias of DynamicTool

get_tool_by_uuid(uuid: UUID | str | None)

get_tool_by_tool_id(tool_id)

get_tool_by_id(object_id)

create_tool(trans, tool_payload, allow_load=True)

list_tools(active=True)

deactivate(dynamic_tool)

foreign_key_name: str

app: BasicSharedApp

class galaxy.managers.tools.ToolFilterMixin

Bases: object

orm_filter_parsers: OrmFilterParsersType

create_tool_filter(attr, op, val)

galaxy.managers.users module

Manager and Serializer for Users.

class galaxy.managers.users.UserManager(app: BasicSharedApp, app_type='galaxy')

Bases: ModelManager, PurgableManagerMixin

foreign_key_name: str = 'user'

__init__(app: BasicSharedApp, app_type='galaxy')

model_class: Type[U]

register(trans, email=None, username=None, password=None, confirm=None, subscribe=False)

Register a new user.

create(email=None, username=None, password=None, **kwargs)

Create a new user.

delete(user, flush=True)

Mark the given user deleted.

undelete(user, flush=True)

Remove the deleted flag for the given user.

purge(user, flush=True)

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

by_id(user_id: int) → User

Gets a model by primary id.

by_email(email: str, filters=None, **kwargs) → User | None

Find a user by their email.

by_api_key(api_key: str, sa_session=None)

Find a user by API key.

by_oidc_access_token(access_token: str)

check_bootstrap_admin_api_key(api_key)

is_admin(user: User | None, trans=None) → bool

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.

admins(filters=None, **kwargs)

Return a list of admin Users.

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

Raise an error if user is not an admin.

Raises:

exceptions.AdminRequiredException – if user is not an admin.

is_anonymous(user: User | None) → bool

Return True if user is anonymous.

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

Raise an error if user is anonymous.

get_user_by_identity(identity)

Get user by username or email.

current_user(trans)

user_can_do_run_as(user) → bool

preferences(user)

private_role(user)

sharing_roles(user)

default_permissions(user)

quota(user, total=False, quota_source_label=None)

quota_bytes(user, quota_source_label: str | None = None)

tags_used(user, tag_models=None)

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

change_password(trans, password=None, confirm=None, token=None, id=None, current=None)

Allows to change a user password with a token.

impersonate(trans, user)

send_activation_email(trans, email, username)

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

send_reset_email(trans, payload, **kwd)

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

get_reset_token(trans, email)

send_subscription_email(email)

activate(user)

get_or_create_remote_user(remote_user_email)

Create a remote user with the email remote_user_email and return it

app: BasicSharedApp

class galaxy.managers.users.UserSerializer(app: MinimalManagerApp)

Bases: ModelSerializer, PurgableSerializerMixin

model_manager_class

alias of UserManager

__init__(app: MinimalManagerApp)

Convert a User and associated data to a dictionary representation.

default_view: str | None

add_serializers()

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

serialize_disk_usage(user: User) → List[UserQuotaUsage]

serialize_disk_usage_for(user: User, label: str | None) → UserQuotaUsage

views: Dict[str, List[str]]

serializable_keyset: Set[str]

serializers: Dict[str, Serializer]

app: MinimalManagerApp

class galaxy.managers.users.UserDeserializer(app: MinimalManagerApp, **kwargs)

Bases: ModelDeserializer

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

model_manager_class

alias of UserManager

add_deserializers()

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

deserialize_preferred_object_store_id(item: Any, key: Any, val: Any, trans=None, **context)

deserialize_username(item, key, username, trans=None, **context)

app: MinimalManagerApp

deserializers: Dict[str, Deserializer]

deserializable_keyset: Set[str]

class galaxy.managers.users.CurrentUserSerializer(app: MinimalManagerApp)

Bases: UserSerializer

model_manager_class

alias of UserManager

serialize(user, keys, **kwargs)

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

serialize_current_anonymous_user(user, keys, trans=None, **kwargs)

default_view: str | None

views: Dict[str, List[str]]

serializable_keyset: Set[str]

serializers: Dict[str, Serializer]

app: MinimalManagerApp

class galaxy.managers.users.AdminUserFilterParser(app: MinimalManagerApp, **kwargs)

Bases: ModelFilterParser, PurgableFiltersMixin

model_manager_class

alias of UserManager

model_class

alias of User

orm_filter_parsers: Dict[str, None | Dict[str, Any] | Callable]

fn_filter_parsers: Dict[str, Any]

dictionary containing parsing data for functional filters - applied after a query is made

app: MinimalManagerApp

galaxy.managers.users.username_from_email(session, email, model_class=<class 'galaxy.model.User'>)

Get next available username generated based on email

galaxy.managers.users.filter_out_invalid_username_characters(username)

Replace invalid characters in username

galaxy.managers.users.username_exists(session, username: str, model_class=<class 'galaxy.model.User'>)

galaxy.managers.users.generate_next_available_username(session, username, model_class=<class 'galaxy.model.User'>)

Generate unique username; user can change it later

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: MinimalManagerApp)

Bases: SharableModelManager

Handle operations outside and between visualizations and other models.

model_class

alias of Visualization

foreign_key_name: str = 'visualization'

user_share_model

alias of VisualizationUserShareAssociation

tag_assoc

alias of VisualizationTagAssociation

annotation_assoc

alias of VisualizationAnnotationAssociation

rating_assoc

alias of VisualizationRatingAssociation

index_query(trans: ProvidesUserContext, payload: VisualizationIndexQueryPayload, include_total_count: bool = False) → Tuple[List[Visualization], int]

app: BasicSharedApp

class galaxy.managers.visualizations.VisualizationSerializer(app: MinimalManagerApp)

Bases: SharableModelSerializer

Interface/service object for serializing visualizations into dictionaries.

model_manager_class

alias of VisualizationManager

SINGLE_CHAR_ABBR: str | None = 'v'

__init__(app: MinimalManagerApp)

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

default_view: str | None

add_serializers()

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

views: Dict[str, List[str]]

serializable_keyset: Set[str]

serializers: Dict[str, Serializer]

app: MinimalManagerApp

class galaxy.managers.visualizations.VisualizationDeserializer(app)

Bases: SharableModelDeserializer

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

model_manager_class

alias of VisualizationManager

__init__(app)

Set up deserializers and validator.

add_deserializers()

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

app: MinimalManagerApp

deserializers: Dict[str, Deserializer]

deserializable_keyset: Set[str]

tag_handler: GalaxyTagHandler

galaxy.managers.visualizations.get_count(session, statement)

galaxy.managers.workflows module

class galaxy.managers.workflows.WorkflowsManager(app: MinimalManagerApp)

Bases: SharableModelManager, DeletableManagerMixin

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

model_class

alias of StoredWorkflow

foreign_key_name: str = 'stored_workflow'

user_share_model

alias of StoredWorkflowUserShareAssociation

__init__(app: MinimalManagerApp)

app: BasicSharedApp

index_query(trans: ProvidesUserContext, payload: WorkflowIndexQueryPayload, include_total_count: bool = False) → Tuple[Result, int | None]

get_stored_workflow(trans, workflow_id, by_stored_id=True) → StoredWorkflow

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

get_stored_accessible_workflow(trans, workflow_id, by_stored_id=True)

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

attach_stored_workflow(trans, workflow)

Attach and return stored workflow if possible.

get_owned_workflow(trans, encoded_workflow_id)

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

check_security(trans, has_workflow, check_ownership=True, check_accessible=True)

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

get_workflow_svg_from_id(trans, id, version=None, for_embed=False) → bytes

get_workflow_svg(trans, workflow, for_embed=False) → bytes

get_invocation(trans, decoded_invocation_id: int, eager=False, check_ownership=True, check_accessible=True) → WorkflowInvocation

get_invocation_report(trans, invocation_id, **kwd)

request_invocation_cancellation(trans, decoded_invocation_id: int)

get_invocation_step(trans, decoded_workflow_invocation_step_id, check_ownership=True, check_accessible=True)

update_invocation_step(trans, decoded_workflow_invocation_step_id, action)

build_invocations_query(trans: ProvidesUserContext, 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, include_nested_invocations=True, check_ownership=True) → Tuple[List, int]

Get invocations owned by the current user.

annotation_assoc: type

class of AnnotationAssociation (e.g. HistoryAnnotationAssociation)

rating_assoc: Type[ItemRatingAssociation]

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

Bases: tuple

stored_workflow: StoredWorkflow

Alias for field number 0

workflow: Workflow

Alias for field number 1

missing_tools: List[Tuple[str, str, str | None, str]]

Alias for field number 2

class galaxy.managers.workflows.WorkflowSerializer(app: MinimalManagerApp)

Bases: SharableModelSerializer

Interface/service object for serializing stored workflows into dictionaries.

These are used for simple workflow operations - much more detailed serialization operations that allow recovering the workflow in its entirity or rendering it in a workflow editor are available inside the WorkflowContentsManager.

model_manager_class

alias of WorkflowsManager

SINGLE_CHAR_ABBR: str | None = 'w'

__init__(app: MinimalManagerApp)

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

add_serializers()

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

default_view: str | None

views: Dict[str, List[str]]

serializable_keyset: Set[str]

serializers: Dict[str, Serializer]

app: MinimalManagerApp

class galaxy.managers.workflows.WorkflowContentsManager(app: MinimalManagerApp, trs_proxy: TrsProxy)

Bases: UsesAnnotations

__init__(app: MinimalManagerApp, trs_proxy: TrsProxy)

ensure_raw_description(dict_or_raw_description)

read_workflow_from_path(app, user, path, allow_in_directory=None) → Workflow

normalize_workflow_format(trans, as_dict)

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.

build_workflow_from_raw_description(trans, raw_workflow_description, workflow_create_options, source=None, add_to_menu=False, hidden=False, is_subworkflow=False) → CreatedWorkflow

update_workflow_from_raw_description(trans, stored_workflow, raw_workflow_description, workflow_update_options)

workflow_to_dict(trans, stored, style='export', version=None, history=None)

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’.

store_workflow_artifacts(directory, filename_base, workflow, **kwd)

store_workflow_to_path(workflow_path, stored_workflow, workflow, **kwd)

static get_step_map_over(current_step, steps)

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.

do_refactor(trans, stored_workflow, refactor_request)

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

refactor(trans, stored_workflow, refactor_request)

get_all_tools(workflow)

get_or_create_workflow_from_trs(trans: ProvidesUserContext, trs_url: str | None, trs_id: str | None = None, trs_version: str | None = None, trs_server: str | None = None)

create_workflow_from_trs_url(trans: ProvidesUserContext, trs_url: str, trs_server: str | None = None) → StoredWorkflow

get_workflow_by_trs_id_and_version(trs_id: str, trs_version: str, user_id: int | None = None) → StoredWorkflow | None

class galaxy.managers.workflows.RefactorRequest(*, actions: List[AddInputAction | galaxy.workflow.refactor.schema.AddStepAction | galaxy.workflow.refactor.schema.ConnectAction | galaxy.workflow.refactor.schema.DisconnectAction | galaxy.workflow.refactor.schema.ExtractInputAction | galaxy.workflow.refactor.schema.ExtractUntypedParameter | galaxy.workflow.refactor.schema.FileDefaultsAction | galaxy.workflow.refactor.schema.FillStepDefaultsAction | galaxy.workflow.refactor.schema.UpdateAnnotationAction | galaxy.workflow.refactor.schema.UpdateCreatorAction | galaxy.workflow.refactor.schema.UpdateNameAction | galaxy.workflow.refactor.schema.UpdateLicenseAction | galaxy.workflow.refactor.schema.UpdateOutputLabelAction | galaxy.workflow.refactor.schema.UpdateReportAction | galaxy.workflow.refactor.schema.UpdateStepLabelAction | galaxy.workflow.refactor.schema.UpdateStepPositionAction | galaxy.workflow.refactor.schema.UpgradeSubworkflowAction | galaxy.workflow.refactor.schema.UpgradeToolAction | galaxy.workflow.refactor.schema.UpgradeAllStepsAction | galaxy.workflow.refactor.schema.RemoveUnlabeledWorkflowOutputs[AddInputAction | AddStepAction | ConnectAction | DisconnectAction | ExtractInputAction | ExtractUntypedParameter | FileDefaultsAction | FillStepDefaultsAction | UpdateAnnotationAction | UpdateCreatorAction | UpdateNameAction | UpdateLicenseAction | UpdateOutputLabelAction | UpdateReportAction | UpdateStepLabelAction | UpdateStepPositionAction | UpgradeSubworkflowAction | UpgradeToolAction | UpgradeAllStepsAction | RemoveUnlabeledWorkflowOutputs]], dry_run: bool = False, style: str = 'export')

Bases: RefactorActions

style: str

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'actions': FieldInfo(annotation=List[Annotated[Union[galaxy.workflow.refactor.schema.AddInputAction, galaxy.workflow.refactor.schema.AddStepAction, galaxy.workflow.refactor.schema.ConnectAction, galaxy.workflow.refactor.schema.DisconnectAction, galaxy.workflow.refactor.schema.ExtractInputAction, galaxy.workflow.refactor.schema.ExtractUntypedParameter, galaxy.workflow.refactor.schema.FileDefaultsAction, galaxy.workflow.refactor.schema.FillStepDefaultsAction, galaxy.workflow.refactor.schema.UpdateAnnotationAction, galaxy.workflow.refactor.schema.UpdateCreatorAction, galaxy.workflow.refactor.schema.UpdateNameAction, galaxy.workflow.refactor.schema.UpdateLicenseAction, galaxy.workflow.refactor.schema.UpdateOutputLabelAction, galaxy.workflow.refactor.schema.UpdateReportAction, galaxy.workflow.refactor.schema.UpdateStepLabelAction, galaxy.workflow.refactor.schema.UpdateStepPositionAction, galaxy.workflow.refactor.schema.UpgradeSubworkflowAction, galaxy.workflow.refactor.schema.UpgradeToolAction, galaxy.workflow.refactor.schema.UpgradeAllStepsAction, galaxy.workflow.refactor.schema.RemoveUnlabeledWorkflowOutputs], FieldInfo(annotation=NoneType, required=True, discriminator='action_type')]], required=True), 'dry_run': FieldInfo(annotation=bool, required=False, default=False), 'style': FieldInfo(annotation=str, required=False, default='export')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

galaxy.managers.workflows.safe_wraps(v: Any, nxt: SerializerFunctionWrapHandler) → str

class galaxy.managers.workflows.RefactorResponse(*, action_executions: List[RefactorActionExecution], workflow: dict[dict], dry_run: bool)

Bases: BaseModel

action_executions: List[RefactorActionExecution]

workflow: dict[dict]

dry_run: bool

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'action_executions': FieldInfo(annotation=List[galaxy.workflow.refactor.schema.RefactorActionExecution], required=True), 'dry_run': FieldInfo(annotation=bool, required=True), 'workflow': FieldInfo(annotation=dict, required=True, metadata=[WrapSerializer(func=<function safe_wraps>, return_type=PydanticUndefined, when_used='json')])}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class galaxy.managers.workflows.WorkflowStateResolutionOptions(*, fill_defaults: bool = False, from_tool_form: bool = False, exact_tools: bool = True)

Bases: BaseModel

fill_defaults: bool

from_tool_form: bool

exact_tools: bool

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'exact_tools': FieldInfo(annotation=bool, required=False, default=True), 'fill_defaults': FieldInfo(annotation=bool, required=False, default=False), 'from_tool_form': FieldInfo(annotation=bool, required=False, default=False)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class galaxy.managers.workflows.WorkflowUpdateOptions(*, fill_defaults: bool = False, from_tool_form: bool = False, exact_tools: bool = True, update_stored_workflow_attributes: bool = True, allow_missing_tools: bool = False, dry_run: bool = False)

Bases: WorkflowStateResolutionOptions

update_stored_workflow_attributes: bool

allow_missing_tools: bool

dry_run: bool

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'allow_missing_tools': FieldInfo(annotation=bool, required=False, default=False), 'dry_run': FieldInfo(annotation=bool, required=False, default=False), 'exact_tools': FieldInfo(annotation=bool, required=False, default=True), 'fill_defaults': FieldInfo(annotation=bool, required=False, default=False), 'from_tool_form': FieldInfo(annotation=bool, required=False, default=False), 'update_stored_workflow_attributes': FieldInfo(annotation=bool, required=False, default=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class galaxy.managers.workflows.WorkflowCreateOptions(*, fill_defaults: bool = False, from_tool_form: bool = False, exact_tools: bool = True, import_tools: bool = False, publish: bool = False, importable: bool | None = None, install_repository_dependencies: bool = False, install_resolver_dependencies: bool = False, install_tool_dependencies: bool = False, new_tool_panel_section_label: str = '', tool_panel_section_id: str = '', tool_panel_section_mapping: Dict = {}, shed_tool_conf: str | None = None, archive_source: str | None = None, trs_tool_id: str | None = None, trs_version_id: str | None = None, trs_server: str | None = None, trs_url: str | None = None)

Bases: WorkflowStateResolutionOptions

import_tools: bool

publish: bool

importable: bool | None

install_repository_dependencies: bool

install_resolver_dependencies: bool

install_tool_dependencies: bool

new_tool_panel_section_label: str

tool_panel_section_id: str

tool_panel_section_mapping: Dict

shed_tool_conf: str | None

archive_source: str | None

trs_tool_id: str | None

trs_version_id: str | None

trs_server: str | None

trs_url: str | None

property is_importable

property install_options

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'archive_source': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'exact_tools': FieldInfo(annotation=bool, required=False, default=True), 'fill_defaults': FieldInfo(annotation=bool, required=False, default=False), 'from_tool_form': FieldInfo(annotation=bool, required=False, default=False), 'import_tools': FieldInfo(annotation=bool, required=False, default=False), 'importable': FieldInfo(annotation=Union[bool, NoneType], required=False, default=None), 'install_repository_dependencies': FieldInfo(annotation=bool, required=False, default=False), 'install_resolver_dependencies': FieldInfo(annotation=bool, required=False, default=False), 'install_tool_dependencies': FieldInfo(annotation=bool, required=False, default=False), 'new_tool_panel_section_label': FieldInfo(annotation=str, required=False, default=''), 'publish': FieldInfo(annotation=bool, required=False, default=False), 'shed_tool_conf': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'tool_panel_section_id': FieldInfo(annotation=str, required=False, default=''), 'tool_panel_section_mapping': FieldInfo(annotation=Dict, required=False, default={}), 'trs_server': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'trs_tool_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'trs_url': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'trs_version_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

exception galaxy.managers.workflows.MissingToolsException(workflow, errors)

Bases: MessageException

__init__(workflow, errors)

class galaxy.managers.workflows.RawWorkflowDescription(as_dict, workflow_path=None)

Bases: object

__init__(as_dict, workflow_path=None)

class galaxy.managers.workflows.Format2ConverterGalaxyInterface

Bases: ImporterGalaxyInterface

import_workflow(workflow, **kwds)

Import a workflow via POST /api/workflows or comparable interface into Galaxy.

galaxy.managers.workflows.get_count(session, statement)