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.item_attrs

import logging

from sqlalchemy.sql.expression import func

# Cannot import galaxy.model b/c it creates a circular import graph.
import galaxy

log = logging.getLogger(__name__)


[docs]class RuntimeException(Exception): pass
[docs]class UsesItemRatings: """ Mixin for getting and setting item ratings. Class makes two assumptions: (1) item-rating association table is named <item_class>RatingAssocation (2) item-rating association table has a column with a foreign key referencing item table that contains the item's id. """
[docs] def get_ave_item_rating_data(self, db_session, item, webapp_model=None): """ Returns the average rating for an item.""" if webapp_model is None: webapp_model = galaxy.model item_rating_assoc_class = self._get_item_rating_assoc_class(item, webapp_model=webapp_model) if not item_rating_assoc_class: raise RuntimeException("Item does not have ratings: %s" % item.__class__.__name__) item_id_filter = self._get_item_id_filter_str(item, item_rating_assoc_class) ave_rating = db_session.query(func.avg(item_rating_assoc_class.rating)).filter(item_id_filter).scalar() # Convert ave_rating to float; note: if there are no item ratings, ave rating is None. if ave_rating: ave_rating = float(ave_rating) else: ave_rating = 0 num_ratings = int(db_session.query(func.count(item_rating_assoc_class.rating)).filter(item_id_filter).scalar()) return (ave_rating, num_ratings)
[docs] def rate_item(self, db_session, user, item, rating, webapp_model=None): """ Rate an item. Return type is <item_class>RatingAssociation. """ if webapp_model is None: webapp_model = galaxy.model item_rating = self.get_user_item_rating(db_session, user, item, webapp_model=webapp_model) if not item_rating: # User has not yet rated item; create rating. item_rating_assoc_class = self._get_item_rating_assoc_class(item, webapp_model=webapp_model) item_rating = item_rating_assoc_class() item_rating.user = user item_rating.set_item(item) item_rating.rating = rating db_session.add(item_rating) db_session.flush() elif item_rating.rating != rating: # User has rated item; update rating. item_rating.rating = rating db_session.flush() return item_rating
[docs] def get_user_item_rating(self, db_session, user, item, webapp_model=None): """ Returns user's rating for an item. Return type is <item_class>RatingAssociation. """ if webapp_model is None: webapp_model = galaxy.model item_rating_assoc_class = self._get_item_rating_assoc_class(item, webapp_model=webapp_model) if not item_rating_assoc_class: raise RuntimeException("Item does not have ratings: %s" % item.__class__.__name__) # Query rating table by user and item id. item_id_filter = self._get_item_id_filter_str(item, item_rating_assoc_class) return db_session.query(item_rating_assoc_class).filter_by(user=user).filter(item_id_filter).first()
def _get_item_rating_assoc_class(self, item, webapp_model=None): """ Returns an item's item-rating association class. """ if webapp_model is None: webapp_model = galaxy.model item_rating_assoc_class = '%sRatingAssociation' % item.__class__.__name__ return getattr(webapp_model, item_rating_assoc_class, None) def _get_item_id_filter_str(self, item, item_rating_assoc_class, webapp_model=None): # Get foreign key in item-rating association table that references item table. if webapp_model is None: webapp_model = galaxy.model item_fk = None for fk in item_rating_assoc_class.table.foreign_keys: if fk.references(item.table): item_fk = fk break if not item_fk: raise RuntimeException("Cannot find item id column in item-rating association table: %s, %s" % item_rating_assoc_class.__name__, item_rating_assoc_class.table.name) # TODO: can we provide a better filter than a raw string? return "%s=%i" % (item_fk.parent.name, item.id)
[docs]class UsesAnnotations: """ Mixin for getting and setting item annotations. """
[docs] def get_item_annotation_str(self, db_session, user, item): """ Returns a user's annotation string for an item. """ if hasattr(item, 'annotations'): # If we already have an annotations object we use it. annotation_obj = None for annotation in item.annotations: if annotation.user == user: annotation_obj = annotation break else: annotation_obj = self.get_item_annotation_obj(db_session, user, item) if annotation_obj: return galaxy.util.unicodify(annotation_obj.annotation) return None
[docs] def get_item_annotation_obj(self, db_session, user, item): """ Returns a user's annotation object for an item. """ # Get annotation association class. annotation_assoc_class = self._get_annotation_assoc_class(item) if not annotation_assoc_class: return None # Get annotation association object. annotation_assoc = db_session.query(annotation_assoc_class).filter_by(user=user) # TODO: use filtering like that in _get_item_id_filter_str() if item.__class__ == galaxy.model.History: annotation_assoc = annotation_assoc.filter_by(history=item) elif item.__class__ == galaxy.model.HistoryDatasetAssociation: annotation_assoc = annotation_assoc.filter_by(hda=item) elif item.__class__ == galaxy.model.StoredWorkflow: annotation_assoc = annotation_assoc.filter_by(stored_workflow=item) elif item.__class__ == galaxy.model.WorkflowStep: annotation_assoc = annotation_assoc.filter_by(workflow_step=item) elif item.__class__ == galaxy.model.Page: annotation_assoc = annotation_assoc.filter_by(page=item) elif item.__class__ == galaxy.model.Visualization: annotation_assoc = annotation_assoc.filter_by(visualization=item) return annotation_assoc.first()
[docs] def add_item_annotation(self, db_session, user, item, annotation): """ Add or update an item's annotation; a user can only have a single annotation for an item. """ # Get/create annotation association object. annotation_assoc = self.get_item_annotation_obj(db_session, user, item) if not annotation_assoc: annotation_assoc_class = self._get_annotation_assoc_class(item) if not annotation_assoc_class: return None annotation_assoc = annotation_assoc_class() item.annotations.append(annotation_assoc) annotation_assoc.user = user # Set annotation. annotation_assoc.annotation = annotation return annotation_assoc
[docs] def delete_item_annotation(self, db_session, user, item): annotation_assoc = self.get_item_annotation_obj(db_session, user, item) if annotation_assoc: db_session.delete(annotation_assoc) db_session.flush()
[docs] def copy_item_annotation(self, db_session, source_user, source_item, target_user, target_item): """ Copy an annotation from a user/item source to a user/item target. """ if source_user and target_user: annotation_str = self.get_item_annotation_str(db_session, source_user, source_item) if annotation_str: annotation = self.add_item_annotation(db_session, target_user, target_item, annotation_str) return annotation return None
def _get_annotation_assoc_class(self, item): """ Returns an item's item-annotation association class. """ class_name = '%sAnnotationAssociation' % item.__class__.__name__ return getattr(galaxy.model, class_name, None)
__all__ = ( 'UsesAnnotations', 'UsesItemRatings', 'RuntimeException', )