Warning

This document is for an old release 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.webapp.controllers.admin

import logging

import tool_shed.grids.admin_grids as admin_grids
from galaxy import (
    util,
    web
)
from galaxy.util import inflector
from galaxy.web.legacy_framework import grids
from galaxy.webapps.base.controller import BaseUIController
from tool_shed.metadata import repository_metadata_manager
from tool_shed.util import (
    metadata_util,
    repository_util,
    shed_util_common as suc
)
from tool_shed.util.admin_util import Admin
from tool_shed.util.web_util import escape

log = logging.getLogger(__name__)


[docs]class AdminController(BaseUIController, Admin): user_list_grid = admin_grids.UserGrid() role_list_grid = admin_grids.RoleGrid() group_list_grid = admin_grids.GroupGrid() manage_category_grid = admin_grids.ManageCategoryGrid() repository_grid = admin_grids.AdminRepositoryGrid() repository_metadata_grid = admin_grids.RepositoryMetadataGrid() delete_operation = grids.GridOperation("Delete", condition=(lambda item: not item.deleted), allow_multiple=True) undelete_operation = grids.GridOperation("Undelete", condition=(lambda item: item.deleted and not item.purged), allow_multiple=True) purge_operation = grids.GridOperation("Purge", condition=(lambda item: item.deleted and not item.purged), allow_multiple=True)
[docs] @web.expose @web.require_admin def browse_repositories(self, trans, **kwd): # We add parameters to the keyword dict in this method in order to rename the param # with an "f-" prefix, simulating filtering by clicking a search link. We have # to take this approach because the "-" character is illegal in HTTP requests. if 'operation' in kwd: operation = kwd['operation'].lower() if operation == "view_or_manage_repository": return trans.response.send_redirect(web.url_for(controller='repository', action='browse_repositories', **kwd)) elif operation == "edit_repository": return trans.response.send_redirect(web.url_for(controller='repository', action='edit_repository', **kwd)) elif operation == "repositories_by_user": # Eliminate the current filters if any exist. for k, v in list(kwd.items()): if k.startswith('f-'): del kwd[k] if 'user_id' in kwd: user = suc.get_user(trans.app, kwd['user_id']) kwd['f-email'] = user.email del kwd['user_id'] else: # The received id is the repository id, so we need to get the id of the user # that uploaded the repository. repository_id = kwd.get('id', None) repository = repository_util.get_repository_in_tool_shed(trans.app, repository_id) kwd['f-email'] = repository.user.email elif operation == "repositories_by_category": # Eliminate the current filters if any exist. for k, v in list(kwd.items()): if k.startswith('f-'): del kwd[k] category_id = kwd.get('id', None) category = suc.get_category(trans.app, category_id) kwd['f-Category.name'] = category.name elif operation == "receive email alerts": if kwd['id']: kwd['caller'] = 'browse_repositories' return trans.response.send_redirect(web.url_for(controller='repository', action='set_email_alerts', **kwd)) else: del kwd['operation'] elif operation == 'delete': return self.delete_repository(trans, **kwd) elif operation == "undelete": return self.undelete_repository(trans, **kwd) # The changeset_revision_select_field in the RepositoryGrid performs a refresh_on_change # which sends in request parameters like changeset_revison_1, changeset_revision_2, etc. One # of the many select fields on the grid performed the refresh_on_change, so we loop through # all of the received values to see which value is not the repository tip. If we find it, we # know the refresh_on_change occurred, and we have the necessary repository id and change set # revision to pass on. for k, v in kwd.items(): changeset_revision_str = 'changeset_revision_' if k.startswith(changeset_revision_str): repository_id = trans.security.encode_id(int(k.lstrip(changeset_revision_str))) repository = repository_util.get_repository_in_tool_shed(trans.app, repository_id) if repository.tip() != v: return trans.response.send_redirect(web.url_for(controller='repository', action='browse_repositories', operation='view_or_manage_repository', id=trans.security.encode_id(repository.id), changeset_revision=v)) # Render the list view return self.repository_grid(trans, **kwd)
[docs] @web.expose @web.require_admin def browse_repository_metadata(self, trans, **kwd): if 'operation' in kwd: operation = kwd['operation'].lower() if operation == "delete": return self.delete_repository_metadata(trans, **kwd) if operation == "view_or_manage_repository_revision": # The received id is a RepositoryMetadata object id, so we need to get the # associated Repository and redirect to view_or_manage_repository with the # changeset_revision. repository_metadata = metadata_util.get_repository_metadata_by_id(trans.app, kwd['id']) repository = repository_metadata.repository kwd['id'] = trans.security.encode_id(repository.id) kwd['changeset_revision'] = repository_metadata.changeset_revision kwd['operation'] = 'view_or_manage_repository' return trans.response.send_redirect(web.url_for(controller='repository', action='browse_repositories', **kwd)) return self.repository_metadata_grid(trans, **kwd)
[docs] @web.expose @web.require_admin def create_category(self, trans, **kwd): message = escape(kwd.get('message', '')) status = kwd.get('status', 'done') name = kwd.get('name', '').strip() description = kwd.get('description', '').strip() if kwd.get('create_category_button', False): if not name or not description: message = 'Enter a valid name and a description' status = 'error' elif suc.get_category_by_name(trans.app, name): message = 'A category with that name already exists' status = 'error' else: # Create the category category = trans.app.model.Category(name=name, description=description) trans.sa_session.add(category) trans.sa_session.flush() # Update the Tool Shed's repository registry. trans.app.repository_registry.add_category_entry(category) message = "Category '%s' has been created" % escape(category.name) status = 'done' trans.response.send_redirect(web.url_for(controller='admin', action='manage_categories', message=message, status=status)) return trans.fill_template('/webapps/tool_shed/category/create_category.mako', name=name, description=description, message=message, status=status)
[docs] @web.expose @web.require_admin def delete_repository(self, trans, **kwd): message = escape(kwd.get('message', '')) status = kwd.get('status', 'done') id = kwd.get('id', None) if id: # Deleting multiple items is currently not allowed (allow_multiple=False), so there will only be 1 id. ids = util.listify(id) count = 0 deleted_repositories = "" for repository_id in ids: repository = repository_util.get_repository_in_tool_shed(trans.app, repository_id) if repository: if not repository.deleted: # Mark all installable repository_metadata records as not installable. for repository_metadata in repository.downloadable_revisions: repository_metadata.downloadable = False trans.sa_session.add(repository_metadata) # Mark the repository admin role as deleted. repository_admin_role = repository.admin_role if repository_admin_role is not None: repository_admin_role.deleted = True trans.sa_session.add(repository_admin_role) repository.deleted = True trans.sa_session.add(repository) trans.sa_session.flush() # Update the repository registry. trans.app.repository_registry.remove_entry(repository) count += 1 deleted_repositories += " %s " % repository.name if count: message = "Deleted %d %s: %s" % (count, inflector.cond_plural(len(ids), "repository"), escape(deleted_repositories)) else: message = "All selected repositories were already marked deleted." else: message = "No repository ids received for deleting." status = 'error' trans.response.send_redirect(web.url_for(controller='admin', action='browse_repositories', message=util.sanitize_text(message), status=status))
[docs] @web.expose @web.require_admin def delete_repository_metadata(self, trans, **kwd): message = escape(kwd.get('message', '')) status = kwd.get('status', 'done') id = kwd.get('id', None) if id: ids = util.listify(id) count = 0 for repository_metadata_id in ids: repository_metadata = metadata_util.get_repository_metadata_by_id(trans.app, repository_metadata_id) trans.sa_session.delete(repository_metadata) trans.sa_session.flush() count += 1 if count: message = "Deleted %d repository metadata %s" % (count, inflector.cond_plural(len(ids), "record")) else: message = "No repository metadata ids received for deleting." status = 'error' trans.response.send_redirect(web.url_for(controller='admin', action='browse_repository_metadata', message=util.sanitize_text(message), status=status))
[docs] @web.expose @web.require_admin def edit_category(self, trans, **kwd): '''Handle requests to edit TS category name or description''' message = escape(kwd.get('message', '')) status = kwd.get('status', 'done') id = kwd.get('id', None) if not id: message = "No category ids received for editing" trans.response.send_redirect(web.url_for(controller='admin', action='manage_categories', message=message, status='error')) category = suc.get_category(trans.app, id) original_category_name = str(category.name) original_category_description = str(category.description) if kwd.get('edit_category_button', False): flush_needed = False new_name = kwd.get('name', '').strip() new_description = kwd.get('description', '').strip() if original_category_name != new_name: if not new_name: message = 'Enter a valid name' status = 'error' elif original_category_name != new_name and suc.get_category_by_name(trans.app, new_name): message = 'A category with that name already exists' status = 'error' else: category.name = new_name flush_needed = True if original_category_description != new_description: category.description = new_description if not flush_needed: flush_needed = True if flush_needed: trans.sa_session.add(category) trans.sa_session.flush() if original_category_name != new_name: # Update the Tool Shed's repository registry. trans.app.repository_registry.edit_category_entry(original_category_name, new_name) message = "The information has been saved for category '%s'" % escape(category.name) status = 'done' return trans.response.send_redirect(web.url_for(controller='admin', action='manage_categories', message=message, status=status)) return trans.fill_template('/webapps/tool_shed/category/edit_category.mako', category=category, message=message, status=status)
[docs] @web.expose @web.require_admin def manage_categories(self, trans, **kwd): if 'f-free-text-search' in kwd: # Trick to enable searching repository name, description from the CategoryGrid. # What we've done is rendered the search box for the RepositoryGrid on the grid.mako # template for the CategoryGrid. See ~/templates/webapps/tool_shed/category/grid.mako. # Since we are searching repositories and not categories, redirect to browse_repositories(). return trans.response.send_redirect(web.url_for(controller='admin', action='browse_repositories', **kwd)) if 'operation' in kwd: operation = kwd['operation'].lower() if operation == "create": return trans.response.send_redirect(web.url_for(controller='admin', action='create_category', **kwd)) elif operation == "delete": return trans.response.send_redirect(web.url_for(controller='admin', action='mark_category_deleted', **kwd)) elif operation == "undelete": return trans.response.send_redirect(web.url_for(controller='admin', action='undelete_category', **kwd)) elif operation == "purge": return trans.response.send_redirect(web.url_for(controller='admin', action='purge_category', **kwd)) elif operation == "edit": return trans.response.send_redirect(web.url_for(controller='admin', action='edit_category', **kwd)) return self.manage_category_grid(trans, **kwd)
[docs] @web.expose @web.require_admin def regenerate_statistics(self, trans, **kwd): message = escape(kwd.get('message', '')) status = kwd.get('status', 'done') if 'regenerate_statistics_button' in kwd: trans.app.shed_counter.generate_statistics() message = "Successfully regenerated statistics" return trans.fill_template('/webapps/tool_shed/admin/statistics.mako', message=message, status=status)
[docs] @web.expose @web.require_admin def manage_role_associations(self, trans, **kwd): """Manage users, groups and repositories associated with a role.""" role_id = kwd.get('id', None) role = repository_util.get_role_by_id(trans.app, role_id) # We currently only have a single role associated with a repository, the repository admin role. repository_role_association = role.repositories[0] repository = repository_role_association.repository associations_dict = repository_util.handle_role_associations(trans.app, role, repository, **kwd) in_users = associations_dict.get('in_users', []) out_users = associations_dict.get('out_users', []) in_groups = associations_dict.get('in_groups', []) out_groups = associations_dict.get('out_groups', []) message = associations_dict.get('message', '') status = associations_dict.get('status', 'done') return trans.fill_template('/webapps/tool_shed/role/role.mako', in_admin_controller=True, repository=repository, role=role, in_users=in_users, out_users=out_users, in_groups=in_groups, out_groups=out_groups, message=message, status=status)
[docs] @web.expose @web.require_admin def reset_metadata_on_selected_repositories_in_tool_shed(self, trans, **kwd): rmm = repository_metadata_manager.RepositoryMetadataManager(trans.app, trans.user) if 'reset_metadata_on_selected_repositories_button' in kwd: message, status = rmm.reset_metadata_on_selected_repositories(**kwd) else: message = escape(util.restore_text(kwd.get('message', ''))) status = kwd.get('status', 'done') repositories_select_field = rmm.build_repository_ids_select_field(name='repository_ids', multiple=True, display='checkboxes', my_writable=False) return trans.fill_template('/webapps/tool_shed/common/reset_metadata_on_selected_repositories.mako', repositories_select_field=repositories_select_field, message=message, status=status)
[docs] @web.expose @web.require_admin def undelete_repository(self, trans, **kwd): message = escape(kwd.get('message', '')) id = kwd.get('id', None) if id: # Undeleting multiple items is currently not allowed (allow_multiple=False), so there will only be 1 id. ids = util.listify(id) count = 0 undeleted_repositories = "" for repository_id in ids: repository = repository_util.get_repository_in_tool_shed(trans.app, repository_id) if repository: if repository.deleted: # Inspect all repository_metadata records to determine those that are installable, and mark # them accordingly. for repository_metadata in repository.metadata_revisions: metadata = repository_metadata.metadata if metadata: if metadata_util.is_downloadable(metadata): repository_metadata.downloadable = True trans.sa_session.add(repository_metadata) # Mark the repository admin role as not deleted. repository_admin_role = repository.admin_role if repository_admin_role is not None: repository_admin_role.deleted = False trans.sa_session.add(repository_admin_role) repository.deleted = False trans.sa_session.add(repository) trans.sa_session.flush() if not repository.deprecated: # Update the repository registry. trans.app.repository_registry.add_entry(repository) count += 1 undeleted_repositories += " %s" % repository.name if count: message = "Undeleted %d %s: %s" % (count, inflector.cond_plural(count, "repository"), undeleted_repositories) else: message = "No selected repositories were marked deleted, so they could not be undeleted." else: message = "No repository ids received for undeleting." trans.response.send_redirect(web.url_for(controller='admin', action='browse_repositories', message=util.sanitize_text(message), status='done'))
[docs] @web.expose @web.require_admin def mark_category_deleted(self, trans, **kwd): # TODO: We should probably eliminate the Category.deleted column since it really makes no # sense to mark a category as deleted (category names and descriptions can be changed instead). # If we do this, and the following 2 methods can be eliminated. message = escape(kwd.get('message', '')) id = kwd.get('id', None) if id: ids = util.listify(id) message = "Deleted %d categories: " % len(ids) for category_id in ids: category = suc.get_category(trans.app, category_id) category.deleted = True trans.sa_session.add(category) trans.sa_session.flush() # Update the Tool Shed's repository registry. trans.app.repository_registry.remove_category_entry(category) message += " %s " % escape(category.name) else: message = "No category ids received for deleting." trans.response.send_redirect(web.url_for(controller='admin', action='manage_categories', message=util.sanitize_text(message), status='done'))
[docs] @web.expose @web.require_admin def purge_category(self, trans, **kwd): # This method should only be called for a Category that has previously been deleted. # Purging a deleted Category deletes all of the following from the database: # - RepoitoryCategoryAssociations where category_id == Category.id message = escape(kwd.get('message', '')) id = kwd.get('id', None) if id: ids = util.listify(id) count = 0 purged_categories = "" message = "Purged %d categories: " % len(ids) for category_id in ids: category = suc.get_category(trans.app, category_id) if category.deleted: # Delete RepositoryCategoryAssociations for rca in category.repositories: trans.sa_session.delete(rca) trans.sa_session.flush() purged_categories += " %s " % category.name message = "Purged %d categories: %s" % (count, escape(purged_categories)) else: message = "No category ids received for purging." trans.response.send_redirect(web.url_for(controller='admin', action='manage_categories', message=util.sanitize_text(message), status='done'))
[docs] @web.expose @web.require_admin def undelete_category(self, trans, **kwd): message = escape(kwd.get('message', '')) id = kwd.get('id', None) if id: ids = util.listify(id) count = 0 undeleted_categories = "" for category_id in ids: category = suc.get_category(trans.app, category_id) if category.deleted: category.deleted = False trans.sa_session.add(category) trans.sa_session.flush() # Update the Tool Shed's repository registry. trans.app.repository_registry.add_category_entry(category) count += 1 undeleted_categories += " %s" % category.name message = "Undeleted %d categories: %s" % (count, escape(undeleted_categories)) else: message = "No category ids received for undeleting." trans.response.send_redirect(web.url_for(controller='admin', action='manage_categories', message=util.sanitize_text(message), status='done'))