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

"""
Manager mixins to unify the interface into things that can contain: Datasets
and other (nested) containers.

(e.g. DatasetCollections, Histories, LibraryFolders)
"""
# Histories should be DatasetCollections.
# Libraries should be DatasetCollections.
import logging
import operator
from typing import Optional

import galaxy.exceptions
import galaxy.util
from galaxy import model

log = logging.getLogger(__name__)


# ====
[docs]class ContainerManagerMixin: """ A class that tracks/contains two types of items: 1) some non-container object (such as datasets) 2) other sub-containers nested within this one Levels of nesting are not considered here; In other words, each of the methods below only work on the first level of nesting. """ # TODO: terminology is getting a bit convoluted and silly at this point: rename three public below? # TODO: this should be an open mapping (not just 2) #: the classes that can be contained contained_class: type subcontainer_class: type #: how any contents lists produced are ordered - (string) attribute name to sort on or tuple of attribute names default_order_by: Optional[str] = None # ---- interface
[docs] def contents(self, container): """ Returns both types of contents: filtered and in some order. """ iters = [] iters.append(self.contained(container)) iters.append(self.subcontainers(container)) return galaxy.util.merge_sorted_iterables(self.order_contents_on, *iters)
[docs] def contained(self, container, **kwargs): """ Returns non-container objects. """ return self._filter_contents(container, self.contained_class, **kwargs)
[docs] def subcontainers(self, container, **kwargs): """ Returns only the containers within this one. """ return self._filter_contents(container, self.subcontainer_class, **kwargs)
# ---- private def _filter_contents(self, container, content_class, **kwargs): # TODO: use list (or by_history etc.) container_filter = self._filter_to_contained(container, content_class) query = self.session().query(content_class).filter(container_filter) return query def _get_filter_for_contained(self, container, content_class): raise galaxy.exceptions.NotImplemented('Abstract class') def _content_manager(self, content): raise galaxy.exceptions.NotImplemented('Abstract class')
[docs]class LibraryFolderAsContainerManagerMixin(ContainerManagerMixin): # can contain two types of subcontainer: LibraryFolder, LibraryDatasetCollectionAssociation # has as the top level container: Library contained_class = model.LibraryDataset subcontainer_class = model.LibraryFolder # subcontainer_class = model.LibraryDatasetCollectionAssociation order_contents_on = operator.attrgetter('create_time') def _get_filter_for_contained(self, container, content_class): if content_class == self.subcontainer_class: return self.subcontainer_class.parent == container return self.contained_class.folder == container def _content_manager(self, content): # type snifffing is inevitable if isinstance(content, model.LibraryDataset): return self.lda_manager elif isinstance(content, model.LibraryFolder): return self.folder_manager raise TypeError(f"Unknown contents class: {str(content)}")
[docs]class DatasetCollectionAsContainerManagerMixin(ContainerManagerMixin): # (note: unlike the other collections, dc's wrap both contained and subcontainers in this class) contained_class = model.DatasetCollectionElement subcontainer_class = model.DatasetCollection order_contents_on = operator.attrgetter('element_index') def _get_filter_for_contained(self, container, content_class): return content_class.collection == container def _content_manager(self, content): # type snifffing is inevitable if isinstance(content, model.DatasetCollectionElement): return self.collection_manager elif isinstance(content, model.DatasetCollection): return self.collection_manager raise TypeError(f"Unknown contents class: {str(content)}")