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

"""
Mixins for Annotatable model managers and serializers.
"""

import abc
import logging
from typing import (
    Dict,
    Optional,
)

from sqlalchemy.orm import scoped_session

from .base import (
    Deserializer,
    FunctionFilterParsersType,
    ModelValidator,
    Serializer,
)

log = logging.getLogger(__name__)


# needed to extract this for use in manager *and* serializer, ideally, would use self.manager.annotation
# from serializer, but history_contents has no self.manager
# TODO: fix
def _match_by_user(item, user) -> Optional[str]:
    if not user:
        return None
    for annotation in item.annotations:
        if annotation.user_id == user.id:
            return annotation.annotation
    return None


[docs]class AnnotatableManagerMixin: #: class of AnnotationAssociation (e.g. HistoryAnnotationAssociation) annotation_assoc: type
[docs] @abc.abstractmethod def session(self) -> scoped_session: ...
[docs] def annotation(self, item) -> Optional[str]: """ Return the annotation string made by the `item`'s owner or `None` if there is no annotation. """ # NOTE: only works with sharable (.user) return self._user_annotation(item, item.user)
# TODO: should/do we support multiple, non-owner annotation of items?
[docs] def annotate(self, item, annotation, user=None, flush=True): """ Create a new annotation on `item` or delete the existing if annotation is `None`. """ if not user: return None if annotation is None: self._delete_annotation(item, user, flush=flush) return None annotation_obj = item.add_item_annotation(self.session(), user, item, annotation) if flush: self.session().flush() return annotation_obj
def _user_annotation(self, item, user): return _match_by_user(item, user) def _delete_annotation(self, item, user, flush=True): returned = item.delete_item_annotation(self.session(), user, item) if flush: self.session().flush() return returned
[docs]class AnnotatableSerializerMixin: serializers: Dict[str, Serializer]
[docs] def add_serializers(self): self.serializers["annotation"] = self.serialize_annotation
[docs] def serialize_annotation(self, item, key, user=None, **context): """ Get and serialize an `item`'s annotation. """ annotation = _match_by_user(item, user) return annotation.strip() if annotation else None
[docs]class AnnotatableDeserializerMixin: deserializers: Dict[str, Deserializer]
[docs] def add_deserializers(self): self.deserializers["annotation"] = self.deserialize_annotation
[docs] def deserialize_annotation(self, item, key, val, user=None, **context): """ Make sure `val` is a valid annotation and assign it, deleting any existing if `val` is None. """ val = ModelValidator.nullable_basestring(key, val) return self.manager.annotate(item, val, user=user, flush=False)
# TODO: I'm not entirely convinced this (or tags) are a good idea for filters since they involve a/the user
[docs]class AnnotatableFilterMixin: fn_filter_parsers: FunctionFilterParsersType def _owner_annotation(self, item) -> Optional[str]: """ Get the annotation by the item's owner. """ return _match_by_user(item, item.user)
[docs] def filter_annotation_contains(self, item, val: str) -> bool: """ Test whether `val` is in the owner's annotation. """ owner_annotation = self._owner_annotation(item) if owner_annotation is None: return False return val.lower() in owner_annotation.lower()
def _add_parsers(self): self.fn_filter_parsers.update( { "annotation": { "op": { "has": self.filter_annotation_contains, "contains": self.filter_annotation_contains, }, }, } )