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.hdcas
"""
Manager and Serializer for HDCAs.
HistoryDatasetCollectionAssociations (HDCAs) are datasets contained or created in a
history.
"""
import logging
from galaxy import model
from galaxy.managers import (
annotatable,
base,
deletable,
hdas,
secured,
taggable
)
log = logging.getLogger(__name__)
# TODO: to DatasetCollectionInstanceManager
[docs]class HDCAManager(
base.ModelManager,
secured.AccessibleManagerMixin,
secured.OwnableManagerMixin,
deletable.PurgableManagerMixin,
taggable.TaggableManagerMixin,
annotatable.AnnotatableManagerMixin):
"""
Interface/service object for interacting with HDCAs.
"""
model_class = model.HistoryDatasetCollectionAssociation
foreign_key_name = 'history_dataset_collection_association'
tag_assoc = model.HistoryDatasetCollectionTagAssociation
annotation_assoc = model.HistoryDatasetCollectionAnnotationAssociation
[docs] def __init__(self, app):
"""
Set up and initialize other managers needed by hdcas.
"""
super(HDCAManager, self).__init__(app)
[docs] def map_datasets(self, content, fn, *parents):
"""
Iterate over the datasets of a given collection, recursing into collections, and
calling fn on each dataset.
Uses the same kwargs as `contents` above.
"""
returned = []
# lots of nesting going on within the nesting
collection = content.collection if hasattr(content, 'collection') else content
this_parents = (content, ) + parents
for element in collection.elements:
next_parents = (element, ) + this_parents
if element.is_collection:
processed_list = self.map_datasets(element.child_collection, fn, *next_parents)
returned.extend(processed_list)
else:
processed = fn(element.dataset_instance, *next_parents)
returned.append(processed)
return returned
# TODO: un-stub
# serializers
# -----------------------------------------------------------------------------
[docs]class DCESerializer(base.ModelSerializer):
"""
Serializer for DatasetCollectionElements.
"""
[docs] def __init__(self, app):
super(DCESerializer, self).__init__(app)
self.hda_serializer = hdas.HDASerializer(app)
self.dc_serializer = DCSerializer(app, dce_serializer=self)
self.default_view = 'summary'
self.add_view('summary', [
'id', 'model_class',
'element_index',
'element_identifier',
'element_type',
'object'
])
[docs] def add_serializers(self):
super(DCESerializer, self).add_serializers()
self.serializers.update({
'model_class' : lambda *a, **c: 'DatasetCollectionElement',
'object' : self.serialize_object
})
[docs] def serialize_object(self, item, key, **context):
if item.hda:
return self.hda_serializer.serialize_to_view(item.hda, view='summary', **context)
if item.child_collection:
return self.dc_serializer.serialize_to_view(item.child_collection, view='detailed', **context)
return 'object'
[docs]class DCSerializer(base.ModelSerializer):
"""
Serializer for DatasetCollections.
"""
[docs] def __init__(self, app, dce_serializer=None):
super(DCSerializer, self).__init__(app)
self.dce_serializer = dce_serializer or DCESerializer(app)
self.default_view = 'summary'
self.add_view('summary', [
'id',
'create_time',
'update_time',
'collection_type',
'populated_state',
'populated_state_message',
'element_count',
])
self.add_view('detailed', [
'populated',
'elements',
], include_keys_from='summary')
[docs] def add_serializers(self):
super(DCSerializer, self).add_serializers()
self.serializers.update({
'model_class' : lambda *a, **c: 'DatasetCollection',
'elements' : self.serialize_elements,
})
[docs] def serialize_elements(self, item, key, **context):
returned = []
for element in item.elements:
serialized = self.dce_serializer.serialize_to_view(element, view='summary', **context)
returned.append(serialized)
return returned
[docs]class DCASerializer(base.ModelSerializer):
"""
Base (abstract) Serializer class for HDCAs and LDCAs.
"""
[docs] def __init__(self, app, dce_serializer=None):
super(DCASerializer, self).__init__(app)
self.dce_serializer = dce_serializer or DCESerializer(app)
self.default_view = 'summary'
self.add_view('summary', [
'id',
'create_time', 'update_time',
'collection_type',
'populated_state',
'populated_state_message',
'element_count',
])
self.add_view('detailed', [
'populated',
'elements',
], include_keys_from='summary')
[docs] def add_serializers(self):
super(DCASerializer, self).add_serializers()
# most attributes are (kinda) proxied from DCs - we need a serializer to proxy to
self.dc_serializer = DCSerializer(self.app)
# then set the serializers to point to it for those attrs
collection_keys = [
'create_time',
'update_time',
'collection_type',
'populated',
'populated_state',
'populated_state_message',
'elements',
'element_count',
]
for key in collection_keys:
self.serializers[key] = self._proxy_to_dataset_collection(key=key)
def _proxy_to_dataset_collection(self, serializer=None, key=None):
# dataset_collection associations are (rough) proxies to datasets - access their serializer using this remapping fn
# remapping done by either kwarg key: IOW dataset attr key (e.g. populated_state)
# or by kwarg serializer: a function that's passed in (e.g. elements)
if key:
return lambda i, k, **c: self.dc_serializer.serialize(i.collection, [k], **c)[k]
if serializer:
return lambda i, k, **c: serializer(i.collection, key or k, **c)
raise TypeError('kwarg serializer or key needed')
[docs]class HDCASerializer(
DCASerializer,
taggable.TaggableSerializerMixin,
annotatable.AnnotatableSerializerMixin):
"""
Serializer for HistoryDatasetCollectionAssociations.
"""
[docs] def __init__(self, app):
super(HDCASerializer, self).__init__(app)
self.hdca_manager = HDCAManager(app)
self.default_view = 'summary'
self.add_view('summary', [
'id',
'type_id',
'name',
'history_id', 'hid',
'history_content_type',
'collection_type',
'populated_state',
'populated_state_message',
'element_count',
'job_source_id',
'job_source_type',
'name',
'type_id',
'deleted',
# 'purged',
'visible',
'type', 'url',
'create_time', 'update_time',
'tags', # TODO: detail view only (maybe)
])
self.add_view('detailed', [
'populated',
'elements'
], include_keys_from='summary')
[docs] def add_serializers(self):
super(HDCASerializer, self).add_serializers()
taggable.TaggableSerializerMixin.add_serializers(self)
annotatable.AnnotatableSerializerMixin.add_serializers(self)
self.serializers.update({
'model_class' : lambda *a, **c: self.hdca_manager.model_class.__class__.__name__,
# TODO: remove
'type' : lambda *a, **c: 'collection',
# part of a history and container
'history_id' : self.serialize_id,
'history_content_type' : lambda *a, **c: self.hdca_manager.model_class.content_type,
'type_id' : self.serialize_type_id,
'job_source_id' : self.serialize_id,
'url' : lambda i, k, **c: self.url_for('history_content_typed',
history_id=self.app.security.encode_id(i.history_id),
id=self.app.security.encode_id(i.id),
type=self.hdca_manager.model_class.content_type),
})