mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
Remove unused manager containers code.
This commit is contained in:
@@ -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)}")
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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]
|
||||
|
||||
Reference in New Issue
Block a user