Warning
This document is for an in-development version 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 UsesItemRatings(object):
"""
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 Exception("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 Exception("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):
# Get foreign key in item-rating association table that references item table.
item_fk = get_foreign_key(item_rating_assoc_class, item)
return item_fk.parent == item.id
[docs]class UsesAnnotations(object):
""" Mixin for getting and setting item annotations. """
[docs] def get_item_annotation_str(self, db_session, user, item):
return get_item_annotation_str(db_session, user, item)
[docs] def get_item_annotation_obj(self, db_session, user, item):
return get_item_annotation_obj(db_session, user, item)
[docs] def add_item_annotation(self, db_session, user, item, annotation):
return add_item_annotation(db_session, user, item, annotation)
[docs] def delete_item_annotation(self, db_session, user, item):
annotation_assoc = 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_item_annotation_obj(db_session, user, item):
"""Returns a user's annotation object for an item."""
# Get annotation association class.
annotation_assoc_class = _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)
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.HistoryDatasetCollectionAssociation:
annotation_assoc = annotation_assoc.filter_by(history_dataset_collection=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()
def get_item_annotation_str(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 = get_item_annotation_obj(db_session, user, item)
if annotation_obj:
return galaxy.util.unicodify(annotation_obj.annotation)
return None
def add_item_annotation(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 = get_item_annotation_obj(db_session, user, item)
if not annotation_assoc:
annotation_assoc_class = _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
def _get_annotation_assoc_class(item):
""" Returns an item's item-annotation association class. """
class_name = '%sAnnotationAssociation' % item.__class__.__name__
return getattr(galaxy.model, class_name, None)
[docs]def get_foreign_key(source_class, target_class):
""" Returns foreign key in source class that references target class. """
target_fk = None
for fk in source_class.table.foreign_keys:
if fk.references(target_class.table):
target_fk = fk
break
if not target_fk:
raise Exception("No foreign key found between objects: %s, %s" % source_class.table, target_class.table)
return target_fk
__all__ = (
'get_foreign_key',
'UsesAnnotations',
'UsesItemRatings',
)