Source code for galaxy.managers.tags

from enum import Enum
from typing import Optional

from pydantic import (

from galaxy.managers.context import ProvidesUserContext
from galaxy.model import ItemTagAssociation
from galaxy.model.tags import GalaxyTagHandlerSession
from galaxy.schema.fields import EncodedDatabaseIdField
from galaxy.schema.schema import TagCollection

taggable_item_names = {item: item for item in ItemTagAssociation.associated_item_names}
# This Enum is generated dynamically and mypy can not statically infer it's real type
# so it should be ignored. See:
TaggableItemClass = Enum("TaggableItemClass", taggable_item_names)  # type: ignore[misc]

[docs]class ItemTagsPayload(BaseModel): item_id: EncodedDatabaseIdField = Field( ..., # This field is required title="Item ID", description="The `encoded identifier` of the item whose tags will be updated.", ) item_class: TaggableItemClass = Field( ..., # This field is required title="Item class", description="The name of the class of the item that will be tagged.", ) item_tags: Optional[TagCollection] = Field( default=None, title="Item tags", description="The list of tags that will replace the current tags associated with the item.", )
[docs] class Config: use_enum_values = True
[docs]class TagsManager: """Interface/service object shared by controllers for interacting with tags."""
[docs] def update(self, trans: ProvidesUserContext, payload: ItemTagsPayload) -> None: """Apply a new set of tags to an item; previous tags are deleted.""" tag_handler = GalaxyTagHandlerSession(trans.sa_session) new_tags: Optional[str] = None if payload.item_tags and len(payload.item_tags.__root__) > 0: new_tags = ",".join(payload.item_tags.__root__) item = self._get_item(trans, payload) user = trans.user tag_handler.delete_item_tags(user, item) tag_handler.apply_item_tags(user, item, new_tags) trans.sa_session.flush()
def _get_item(self, trans: ProvidesUserContext, payload: ItemTagsPayload): """ Get an item based on type and id. """ tag_handler = GalaxyTagHandlerSession(trans.sa_session) id = item_class_name = str(payload.item_class) item_class = tag_handler.item_tag_assoc_info[item_class_name].item_class item = trans.sa_session.query(item_class).filter( == id).first() return item