Source code for galaxy.managers.libraries

Manager and Serializer for libraries.
import logging

from sqlalchemy import false, not_, or_, true
from sqlalchemy.orm.exc import MultipleResultsFound
from sqlalchemy.orm.exc import NoResultFound

from galaxy import exceptions
from galaxy.managers import folders
from galaxy.util import (

log = logging.getLogger(__name__)

# =============================================================================
[docs]class LibraryManager(object): """ Interface/service object for interacting with libraries. """
[docs] def __init__(self, *args, **kwargs): super(LibraryManager, self).__init__(*args, **kwargs)
[docs] def get(self, trans, decoded_library_id, check_accessible=True): """ Get the library from the DB. :param decoded_library_id: decoded library id :type decoded_library_id: int :param check_accessible: flag whether to check that user can access item :type check_accessible: bool :returns: the requested library :rtype: galaxy.model.Library """ try: library = trans.sa_session.query( == decoded_library_id).one() except MultipleResultsFound: raise exceptions.InconsistentDatabase('Multiple libraries found with the same id.') except NoResultFound: raise exceptions.RequestParameterInvalidException('No library found with the id provided.') except Exception as e: raise exceptions.InternalServerError('Error loading from the database.' + unicodify(e)) library =, library, check_accessible) return library
[docs] def create(self, trans, name, description='', synopsis=''): """ Create a new library. """ if not trans.user_is_admin: raise exceptions.ItemAccessibilityException('Only administrators can create libraries.') else: library =, description=description, synopsis=synopsis) root_folder =, description='') library.root_folder = root_folder trans.sa_session.add_all((library, root_folder)) trans.sa_session.flush() return library
[docs] def update(self, trans, library, name=None, description=None, synopsis=None): """ Update the given library """ changed = False if not trans.user_is_admin: raise exceptions.ItemAccessibilityException('Only administrators can update libraries.') if library.deleted: raise exceptions.RequestParameterInvalidException('You cannot modify a deleted library. Undelete it first.') if name is not None: = name changed = True # When library is renamed the root folder has to be renamed too. folder_manager = folders.FolderManager() folder_manager.update(trans, library.root_folder, name=name) if description is not None: library.description = description changed = True if synopsis is not None: library.synopsis = synopsis changed = True if changed: trans.sa_session.add(library) trans.sa_session.flush() return library
[docs] def delete(self, trans, library, undelete=False): """ Mark given library deleted/undeleted based on the flag. """ if not trans.user_is_admin: raise exceptions.ItemAccessibilityException('Only administrators can delete and undelete libraries.') if undelete: library.deleted = False else: library.deleted = True trans.sa_session.add(library) trans.sa_session.flush() return library
[docs] def list(self, trans, deleted=False): """ Return a list of libraries from the DB. :param deleted: if True, show only ``deleted`` libraries, if False show only ``non-deleted`` :type deleted: boolean (optional) :returns: query that will emit all accessible libraries :rtype: sqlalchemy query :returns: dict of 3 sets with available actions for user's accessible libraries and a set of ids of all public libraries. These are used for limiting the number of queries when dictifying the libraries later on. :rtype: dict """ is_admin = trans.user_is_admin query = trans.sa_session.query( library_access_action = restricted_library_ids = {lp.library_id for lp in ( trans.sa_session.query(trans.model.LibraryPermissions).filter( trans.model.LibraryPermissions.table.c.action == library_access_action ).distinct())} prefetched_ids = {'restricted_library_ids': restricted_library_ids} if is_admin: if deleted is None: # Flag is not specified, do not filter on it. pass elif deleted: query = query.filter( == true()) else: query = query.filter( == false()) else: # Nonadmins can't see deleted libraries query = query.filter( == false()) current_user_role_ids = [ for role in trans.get_current_user_roles()] all_actions = trans.sa_session.query(trans.model.LibraryPermissions).filter(trans.model.LibraryPermissions.table.c.role_id.in_(current_user_role_ids)) library_add_action = library_modify_action = library_manage_action = accessible_restricted_library_ids = set() allowed_library_add_ids = set() allowed_library_modify_ids = set() allowed_library_manage_ids = set() for action in all_actions: if action.action == library_access_action: accessible_restricted_library_ids.add(action.library_id) if action.action == library_add_action: allowed_library_add_ids.add(action.library_id) if action.action == library_modify_action: allowed_library_modify_ids.add(action.library_id) if action.action == library_manage_action: allowed_library_manage_ids.add(action.library_id) query = query.filter(or_( not_(, )) prefetched_ids['allowed_library_add_ids'] = allowed_library_add_ids prefetched_ids['allowed_library_modify_ids'] = allowed_library_modify_ids prefetched_ids['allowed_library_manage_ids'] = allowed_library_manage_ids return query, prefetched_ids
[docs] def secure(self, trans, library, check_accessible=True): """ Check if library is accessible to user. :param library: library :type library: galaxy.model.Library :param check_accessible: flag whether to check that user can access library :type check_accessible: bool :returns: the original library :rtype: Library """ # all libraries are accessible to an admin if trans.user_is_admin: return library if check_accessible: library = self.check_accessible(trans, library) return library
[docs] def check_accessible(self, trans, library): """ Check whether the library is accessible to current user. """ if not, library): raise exceptions.ObjectNotFound('Library with the id provided was not found.') elif library.deleted: raise exceptions.ObjectNotFound('Library with the id provided is deleted.') else: return library
[docs] def get_library_dict(self, trans, library, prefetched_ids=None): """ Return library data in the form of a dictionary. :param library: library :type library: galaxy.model.Library :param prefetched_ids: dict of 3 sets with available actions for user's accessible libraries and a set of ids of all public libraries. These are used for limiting the number of queries when dictifying a set of libraries. :type prefetched_ids: dict :returns: dict with data about the library :rtype: dictionary """ restricted_library_ids = prefetched_ids.get('restricted_library_ids', None) if prefetched_ids else None allowed_library_add_ids = prefetched_ids.get('allowed_library_add_ids', None) if prefetched_ids else None allowed_library_modify_ids = prefetched_ids.get('allowed_library_modify_ids', None) if prefetched_ids else None allowed_library_manage_ids = prefetched_ids.get('allowed_library_manage_ids', None) if prefetched_ids else None library_dict = library.to_dict(view='element', value_mapper={'id':, 'root_folder_id':}) library_dict['public'] = False if (restricted_library_ids and in restricted_library_ids) else True library_dict['create_time_pretty'] = pretty_print_time_interval(library.create_time, precise=True) if not trans.user_is_admin: if prefetched_ids: library_dict['can_user_add'] = True if (allowed_library_add_ids and in allowed_library_add_ids) else False library_dict['can_user_modify'] = True if (allowed_library_modify_ids and in allowed_library_modify_ids) else False library_dict['can_user_manage'] = True if (allowed_library_manage_ids and in allowed_library_manage_ids) else False else: current_user_roles = trans.get_current_user_roles() library_dict['can_user_add'] =, library) library_dict['can_user_modify'] =, library) library_dict['can_user_manage'] =, library) else: library_dict['can_user_add'] = True library_dict['can_user_modify'] = True library_dict['can_user_manage'] = True return library_dict
[docs] def get_current_roles(self, trans, library): """ Load all permissions currently related to the given library. :param library: the model object :type library: galaxy.model.Library :rtype: dictionary :returns: dict of current roles for all available permission types """ access_library_role_list = [(, for access_role in self.get_access_roles(trans, library)] modify_library_role_list = [(, for modify_role in self.get_modify_roles(trans, library)] manage_library_role_list = [(, for manage_role in self.get_manage_roles(trans, library)] add_library_item_role_list = [(, for add_role in self.get_add_roles(trans, library)] return dict(access_library_role_list=access_library_role_list, modify_library_role_list=modify_library_role_list, manage_library_role_list=manage_library_role_list, add_library_item_role_list=add_library_item_role_list)
[docs] def get_access_roles(self, trans, library): """ Load access roles for all library permissions """ return set(library.get_access_roles(trans))
[docs] def get_modify_roles(self, trans, library): """ Load modify roles for all library permissions """ return set(,
[docs] def get_manage_roles(self, trans, library): """ Load manage roles for all library permissions """ return set(,
[docs] def get_add_roles(self, trans, library): """ Load add roles for all library permissions """ return set(,
[docs] def set_permission_roles(self, trans, library, access_roles, modify_roles, manage_roles, add_roles): """ Set permissions on the given library. """
[docs] def make_public(self, trans, library): """ Makes the given library public (removes all access roles) """ return self.is_public(trans, library)
[docs] def is_public(self, trans, library): """ Return true if lib is public. """ return