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.
        """