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 tool_shed.managers.groups

"""
Manager and Serializer for TS groups.
"""
import logging

from sqlalchemy import (
    false,
    true,
)
from sqlalchemy.orm.exc import (
    MultipleResultsFound,
    NoResultFound,
)

from galaxy.exceptions import (
    Conflict,
    InconsistentDatabase,
    InternalServerError,
    ItemAccessibilityException,
    ObjectNotFound,
    RequestParameterInvalidException,
)

log = logging.getLogger(__name__)


# =============================================================================
[docs]class GroupManager: """ Interface/service object for interacting with TS groups. """
[docs] def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs)
[docs] def get(self, trans, decoded_group_id=None, name=None): """ Get the group from the DB based on its ID or name. :param decoded_group_id: decoded group id :type decoded_group_id: int :returns: the requested group :rtype: tool_shed.model.Group """ if decoded_group_id is None and name is None: raise RequestParameterInvalidException("You must supply either ID or a name of the group.") name_query = trans.sa_session.query(trans.app.model.Group).filter(trans.app.model.Group.table.c.name == name) id_query = trans.sa_session.query(trans.app.model.Group).filter( trans.app.model.Group.table.c.id == decoded_group_id ) try: group = id_query.one() if decoded_group_id else name_query.one() except MultipleResultsFound: raise InconsistentDatabase("Multiple groups found with the same identifier.") except NoResultFound: raise ObjectNotFound("No group found with the identifier provided.") except Exception: raise InternalServerError("Error loading from the database.") return group
[docs] def create(self, trans, name, description=""): """ Create a new group. """ if not trans.user_is_admin: raise ItemAccessibilityException("Only administrators can create groups.") else: if self.get(trans, name=name): raise Conflict(f"Group with the given name already exists. Name: {str(name)}") # TODO add description field to the model group = trans.app.model.Group(name=name) trans.sa_session.add(group) trans.sa_session.flush() return group
[docs] def update(self, trans, group, name=None, description=None): """ Update the given group """ changed = False if not trans.user_is_admin: raise ItemAccessibilityException("Only administrators can update groups.") if group.deleted: raise RequestParameterInvalidException("You cannot modify a deleted group. Undelete it first.") if name is not None: group.name = name changed = True if description is not None: group.description = description changed = True if changed: trans.sa_session.add(group) trans.sa_session.flush() return group
[docs] def delete(self, trans, group, undelete=False): """ Mark given group deleted/undeleted based on the flag. """ if not trans.user_is_admin: raise ItemAccessibilityException("Only administrators can delete and undelete groups.") if undelete: group.deleted = False else: group.deleted = True trans.sa_session.add(group) trans.sa_session.flush() return group
[docs] def list(self, trans, deleted=False): """ Return a list of groups from the DB. :returns: query that will emit all groups :rtype: sqlalchemy query """ is_admin = trans.user_is_admin query = trans.sa_session.query(trans.app.model.Group) if is_admin: if deleted is None: # Flag is not specified, do not filter on it. pass elif deleted: query = query.filter(trans.app.model.Group.table.c.deleted == true()) else: query = query.filter(trans.app.model.Group.table.c.deleted == false()) else: query = query.filter(trans.app.model.Group.table.c.deleted == false()) return query