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.webapps.galaxy.api.annotations
"""
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