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.model.dataset_collections.type_description

from .registry import DATASET_COLLECTION_TYPES_REGISTRY


[docs]class CollectionTypeDescriptionFactory:
[docs] def __init__(self, type_registry=DATASET_COLLECTION_TYPES_REGISTRY): # taking in type_registry though not using it, because we will someday # I think. self.type_registry = type_registry
[docs] def for_collection_type(self, collection_type): assert collection_type is not None return CollectionTypeDescription(collection_type, self)
[docs]class CollectionTypeDescription: """Abstraction over dataset collection type that ties together string reprentation in database/model with type registry. >>> factory = CollectionTypeDescriptionFactory(None) >>> nested_type_description = factory.for_collection_type("list:paired") >>> paired_type_description = factory.for_collection_type("paired") >>> nested_type_description.has_subcollections_of_type("list") False >>> nested_type_description.has_subcollections_of_type("list:paired") False >>> nested_type_description.has_subcollections_of_type("paired") True >>> nested_type_description.has_subcollections_of_type(paired_type_description) True >>> nested_type_description.has_subcollections() True >>> paired_type_description.has_subcollections() False >>> paired_type_description.rank_collection_type() 'paired' >>> nested_type_description.rank_collection_type() 'list' >>> nested_type_description.effective_collection_type(paired_type_description) 'list' >>> nested_type_description.effective_collection_type_description(paired_type_description).collection_type 'list' >>> nested_type_description.child_collection_type() 'paired' """
[docs] def __init__(self, collection_type, collection_type_description_factory): self.collection_type = collection_type self.collection_type_description_factory = collection_type_description_factory self.__has_subcollections = self.collection_type.find(":") > 0
[docs] def child_collection_type(self): rank_collection_type = self.rank_collection_type() return self.collection_type[len(rank_collection_type) + 1 :]
[docs] def child_collection_type_description(self): child_collection_type = self.child_collection_type() return self.collection_type_description_factory.for_collection_type(child_collection_type)
[docs] def effective_collection_type_description(self, subcollection_type): effective_collection_type = self.effective_collection_type(subcollection_type) return self.collection_type_description_factory.for_collection_type(effective_collection_type)
[docs] def effective_collection_type(self, subcollection_type): if hasattr(subcollection_type, "collection_type"): subcollection_type = subcollection_type.collection_type if not self.has_subcollections_of_type(subcollection_type): raise ValueError(f"Cannot compute effective subcollection type of {subcollection_type} over {self}") return self.collection_type[: -(len(subcollection_type) + 1)]
[docs] def has_subcollections_of_type(self, other_collection_type): """Take in another type (either flat string or another CollectionTypeDescription) and determine if this collection contains subcollections matching that type. The way this is used in map/reduce it seems to make the most sense for this to return True if these subtypes are proper (i.e. a type is not considered to have subcollections of its own type). """ if hasattr(other_collection_type, "collection_type"): other_collection_type = other_collection_type.collection_type collection_type = self.collection_type return collection_type.endswith(other_collection_type) and collection_type != other_collection_type
[docs] def is_subcollection_of_type(self, other_collection_type): if not hasattr(other_collection_type, "collection_type"): other_collection_type = self.collection_type_description_factory.for_collection_type(other_collection_type) return other_collection_type.has_subcollections_of_type(self)
[docs] def can_match_type(self, other_collection_type): if hasattr(other_collection_type, "collection_type"): other_collection_type = other_collection_type.collection_type collection_type = self.collection_type return other_collection_type == collection_type
[docs] def subcollection_type_description(self): if not self.__has_subcollections: raise ValueError(f"Cannot generate subcollection type description for flat type {self.collection_type}") subcollection_type = self.collection_type.split(":", 1)[1] return self.collection_type_description_factory.for_collection_type(subcollection_type)
[docs] def has_subcollections(self): return self.__has_subcollections
[docs] def rank_collection_type(self): """Return the top-level collection type corresponding to this collection type. For instance the "rank" type of a list of paired data ("list:paired") is "list". """ return self.collection_type.split(":")[0]
[docs] def rank_type_plugin(self): return self.collection_type_description_factory.type_registry.get(self.rank_collection_type())
@property def dimension(self): return len(self.collection_type.split(":")) + 1
[docs] def multiply(self, other_collection_type): collection_type = map_over_collection_type(self, other_collection_type) return self.collection_type_description_factory.for_collection_type(collection_type)
def __str__(self): return f"CollectionTypeDescription[{self.collection_type}]"
[docs]def map_over_collection_type(mapped_over_collection_type, target_collection_type): if hasattr(mapped_over_collection_type, "collection_type"): mapped_over_collection_type = mapped_over_collection_type.collection_type if not target_collection_type: return mapped_over_collection_type else: if hasattr(target_collection_type, "collection_type"): target_collection_type = target_collection_type.collection_type return f"{mapped_over_collection_type}:{target_collection_type}"
COLLECTION_TYPE_DESCRIPTION_FACTORY = CollectionTypeDescriptionFactory()