diff --git a/lib/galaxy/managers/containers.py b/lib/galaxy/managers/containers.py deleted file mode 100644 index 860fdcf4d23..00000000000 --- a/lib/galaxy/managers/containers.py +++ /dev/null @@ -1,114 +0,0 @@ -""" -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, Type - -import galaxy.exceptions -import galaxy.util -from galaxy import model - -log = logging.getLogger(__name__) - - -# ==== -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[model._HasTable] - subcontainer_class: Type[model._HasTable] - #: 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 - 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) - - def contained(self, container, **kwargs): - """ - Returns non-container objects. - """ - return self._filter_contents(container, self.contained_class, **kwargs) - - 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') - - -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)}") - - -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)}") diff --git a/lib/galaxy/managers/history_contents.py b/lib/galaxy/managers/history_contents.py index 3c0dcbc2072..7766e35d905 100644 --- a/lib/galaxy/managers/history_contents.py +++ b/lib/galaxy/managers/history_contents.py @@ -3,6 +3,7 @@ Heterogenous lists/contents are difficult to query properly since unions are not easily made. """ import logging +from typing import Any, Dict, List from sqlalchemy import ( asc, @@ -25,7 +26,6 @@ from galaxy import ( from galaxy.managers import ( annotatable, base, - containers, deletable, hdas, hdcas, @@ -33,14 +33,14 @@ from galaxy.managers import ( tools ) from galaxy.structured_app import MinimalManagerApp -from .base import raise_filter_err +from .base import raise_filter_err, Serializer log = logging.getLogger(__name__) # into its own class to have it's own filters, etc. # TODO: but can't inherit from model manager (which assumes only one model) -class HistoryContentsManager(containers.ContainerManagerMixin, base.SortableManager): +class HistoryContentsManager(base.SortableManager): root_container_class = model.History @@ -225,7 +225,7 @@ class HistoryContentsManager(containers.ContainerManagerMixin, base.SortableMana return contents_results # partition ids into a map of { component_class names -> list of ids } from the above union query - id_map = dict(((self.contained_class_type_name, []), (self.subcontainer_class_type_name, []))) + id_map: Dict[str, List[int]] = dict([(self.contained_class_type_name, []), (self.subcontainer_class_type_name, [])]) for result in contents_results: result_type = self._get_union_type(result) contents_id = self._get_union_id(result) @@ -344,14 +344,14 @@ class HistoryContentsManager(containers.ContainerManagerMixin, base.SortableMana ) subquery = self._session().query(*columns) # for the HDA's we need to join the Dataset since it has an actual state column - subquery = subquery.join(model.Dataset, model.Dataset.id == component_class.dataset_id) + subquery = subquery.join(model.Dataset, model.Dataset.id == component_class.table.c.dataset_id) if history_id: - subquery = subquery.filter(component_class.history_id == history_id) + subquery = subquery.filter(component_class.table.c.history_id == history_id) else: # Make sure we only return items that are user-accessible by checking that they are in a history # owned by the current user. # TODO: move into filter mixin, and implement accessible logic as SQL query - subquery = subquery.filter(component_class.history_id == model.History.table.c.id, + subquery = subquery.filter(component_class.table.c.history_id == model.History.table.c.id, model.History.table.c.user_id == user_id) return subquery diff --git a/setup.cfg b/setup.cfg index e5339897287..23ce19b9a29 100644 --- a/setup.cfg +++ b/setup.cfg @@ -403,8 +403,6 @@ check_untyped_defs = False check_untyped_defs = False [mypy-galaxy.managers.taggable] check_untyped_defs = False -[mypy-galaxy.managers.containers] -check_untyped_defs = False [mypy-galaxy.managers.collections_util] check_untyped_defs = False [mypy-galaxy.jobs.splitters.multi]