Warning

This document is for an old release 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.

Source code for galaxy.dataset_collections.builder

from galaxy import model
from galaxy.util.odict import odict


[docs]def build_collection(type, dataset_instances): """ Build DatasetCollection with populated DatasetcollectionElement objects corresponding to the supplied dataset instances or throw exception if this is not a valid collection of the specified type. """ dataset_collection = model.DatasetCollection() set_collection_elements(dataset_collection, type, dataset_instances) return dataset_collection
[docs]def set_collection_elements(dataset_collection, type, dataset_instances): element_index = 0 elements = [] for element in type.generate_elements(dataset_instances): element.element_index = element_index element.collection = dataset_collection elements.append(element) element_index += 1 dataset_collection.elements = elements dataset_collection.element_count = element_index return dataset_collection
[docs]class CollectionBuilder(object): """ Purely functional builder pattern for building a dataset collection. """
[docs] def __init__(self, collection_type_description): self._collection_type_description = collection_type_description self._current_elements = odict()
[docs] def get_level(self, identifier): if not self._nested_collection: message_template = "Cannot add nested collection to collection of type [%s]" message = message_template % (self._collection_type_description) raise AssertionError(message) if identifier not in self._current_elements: subcollection_builder = CollectionBuilder( self._subcollection_type_description ) self._current_elements[identifier] = subcollection_builder return self._current_elements[identifier]
[docs] def add_dataset(self, identifier, dataset_instance): self._current_elements[identifier] = dataset_instance
[docs] def build_elements(self): elements = self._current_elements if self._nested_collection: new_elements = odict() for identifier, element in elements.items(): new_elements[identifier] = element.build() elements = new_elements return elements
[docs] def build(self): type_plugin = self._collection_type_description.rank_type_plugin() collection = build_collection(type_plugin, self.build_elements()) collection.collection_type = self._collection_type_description.collection_type return collection
@property def _subcollection_type_description(self): return self._collection_type_description.subcollection_type_description() @property def _nested_collection(self): return self._collection_type_description.has_subcollections()
[docs]class BoundCollectionBuilder(CollectionBuilder): """ More stateful builder that is bound to a particular model object. """
[docs] def __init__(self, dataset_collection, collection_type_description): self.dataset_collection = dataset_collection if dataset_collection.populated: raise Exception("Cannot reset elements of an already populated dataset collection.") super(BoundCollectionBuilder, self).__init__(collection_type_description)
[docs] def populate(self): elements = self.build_elements() type_plugin = self._collection_type_description.rank_type_plugin() set_collection_elements(self.dataset_collection, type_plugin, elements) self.dataset_collection.mark_as_populated()