"""
Mixins for Annotatable model managers and serializers.
"""
import abc
import logging
from typing import (
Dict,
Optional,
)
from sqlalchemy.orm import scoped_session
from galaxy.model.base import transaction
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:
session = self.session()
with transaction(session):
session.commit()
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:
session = self.session()
with transaction(session):
session.commit()
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,
},
},
}
)