Source code for tool_shed.managers.repositories

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

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

from galaxy.exceptions import (
    InconsistentDatabase,
    InternalServerError,
    RequestParameterInvalidException,
)

log = logging.getLogger(__name__)


# =============================================================================
[docs]class RepoManager: """ Interface/service object for interacting with TS repositories. """
[docs] def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs)
[docs] def get(self, trans, decoded_repo_id): """ Get the repo from the DB. :param decoded_repo_id: decoded repo id :type decoded_repo_id: int :returns: the requested repo :rtype: tool_shed.webapp.model.Repository """ try: repo = ( trans.sa_session.query(trans.app.model.Repository) .filter(trans.app.model.Repository.table.c.id == decoded_repo_id) .one() ) except MultipleResultsFound: raise InconsistentDatabase("Multiple repositories found with the same id.") except NoResultFound: raise RequestParameterInvalidException("No repository found with the id provided.") except Exception: raise InternalServerError("Error loading from the database.") return repo
[docs] def list_by_owner(self, trans, user_id): """ Return a list of of repositories owned by a given TS user from the DB. :returns: query that will emit repositories owned by given user :rtype: sqlalchemy query """ query = trans.sa_session.query(trans.app.model.Repository).filter( trans.app.model.Repository.table.c.user_id == user_id ) return query
[docs] def create(self, trans, name, description=""): """ Create a new group. """
[docs] def update(self, trans, group, name=None, description=None): """ Update the given group """
[docs] def delete(self, trans, group, undelete=False): """ Mark given group deleted/undeleted based on the flag. """