"""
API operations on annotations.
"""
import logging
from abc import abstractmethod
from galaxy import (
exceptions,
managers,
)
from galaxy.managers.context import ProvidesHistoryContext
from galaxy.model.base import transaction
from galaxy.model.item_attrs import UsesAnnotations
from galaxy.util.sanitize_html import sanitize_html
from galaxy.web import expose_api
from galaxy.webapps.base.controller import UsesStoredWorkflowMixin
from . import (
BaseGalaxyAPIController,
depends,
)
log = logging.getLogger(__name__)
[docs]class BaseAnnotationsController(BaseGalaxyAPIController, UsesStoredWorkflowMixin, UsesAnnotations):
tagged_item_id: str
[docs] @expose_api
def index(self, trans: ProvidesHistoryContext, **kwd):
idnum = kwd[self.tagged_item_id]
if (item := self._get_item_from_id(trans, idnum)) is not None:
return self.get_item_annotation_str(trans.sa_session, trans.user, item)
[docs] @expose_api
def create(self, trans: ProvidesHistoryContext, payload: dict, **kwd):
if "text" not in payload:
return ""
idnum = kwd[self.tagged_item_id]
if (item := self._get_item_from_id(trans, idnum)) is not None:
new_annotation = payload.get("text")
# TODO: sanitize on display not entry
new_annotation = sanitize_html(new_annotation)
self.add_item_annotation(trans.sa_session, trans.user, item, new_annotation)
with transaction(trans.sa_session):
trans.sa_session.commit()
return new_annotation
return ""
[docs] @expose_api
def delete(self, trans: ProvidesHistoryContext, **kwd):
idnum = kwd[self.tagged_item_id]
if (item := self._get_item_from_id(trans, idnum)) is not None:
return self.delete_item_annotation(trans.sa_session, trans.user, item)
[docs] @expose_api
def undelete(self, trans: ProvidesHistoryContext, **kwd):
raise exceptions.NotImplemented()
@abstractmethod
def _get_item_from_id(self, trans: ProvidesHistoryContext, idstr):
"""Return item with annotation association."""
[docs]class HistoryAnnotationsController(BaseAnnotationsController):
controller_name = "history_annotations"
tagged_item_id = "history_id"
history_manager: managers.histories.HistoryManager = depends(managers.histories.HistoryManager)
def _get_item_from_id(self, trans: ProvidesHistoryContext, idstr):
decoded_idstr = self.decode_id(idstr)
history = self.history_manager.get_accessible(decoded_idstr, trans.user, current_history=trans.history)
return history
[docs]class HistoryContentAnnotationsController(BaseAnnotationsController):
controller_name = "history_content_annotations"
tagged_item_id = "history_content_id"
hda_manager: managers.hdas.HDAManager = depends(managers.hdas.HDAManager)
def _get_item_from_id(self, trans: ProvidesHistoryContext, idstr):
decoded_idstr = self.decode_id(idstr)
hda = self.hda_manager.get_accessible(decoded_idstr, trans.user)
hda = self.hda_manager.error_if_uploading(hda)
return hda
[docs]class WorkflowAnnotationsController(BaseAnnotationsController):
controller_name = "workflow_annotations"
tagged_item_id = "workflow_id"
def _get_item_from_id(self, trans: ProvidesHistoryContext, idstr):
hda = self.get_stored_workflow(trans, idstr)
return hda