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,
)
from galaxy.model.base import transaction
log = logging.getLogger(__name__)
# =============================================================================
[docs]class GroupManager:
"""
Interface/service object for interacting with TS groups.
"""
[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)
with transaction(trans.sa_session):
trans.sa_session.commit()
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)
with transaction(trans.sa_session):
trans.sa_session.commit()
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)
with transaction(trans.sa_session):
trans.sa_session.commit()
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