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 galaxy.actions.admin

"""
Contains administrative functions
"""
import logging

from galaxy import util
from galaxy.exceptions import ActionInputError

log = logging.getLogger(__name__)


[docs]class AdminActions(object): """ Mixin for controllers that provide administrative functionality. """ def _create_quota(self, params, decode_id=None): if params.amount.lower() in ('unlimited', 'none', 'no limit'): create_amount = None else: try: create_amount = util.size_to_bytes(params.amount) except AssertionError: create_amount = False if not params.name or not params.description: raise ActionInputError("Enter a valid name and a description.") elif self.sa_session.query(self.app.model.Quota).filter(self.app.model.Quota.table.c.name == params.name).first(): raise ActionInputError("Quota names must be unique and a quota with that name already exists, so choose another name.") elif not params.get('amount', None): raise ActionInputError("Enter a valid quota amount.") elif create_amount is False: raise ActionInputError("Unable to parse the provided amount.") elif params.operation not in self.app.model.Quota.valid_operations: raise ActionInputError("Enter a valid operation.") elif params.default != 'no' and params.default not in self.app.model.DefaultQuotaAssociation.types.__dict__.values(): raise ActionInputError("Enter a valid default type.") elif params.default != 'no' and params.operation != '=': raise ActionInputError("Operation for a default quota must be '='.") elif create_amount is None and params.operation != '=': raise ActionInputError("Operation for an unlimited quota must be '='.") else: # Create the quota quota = self.app.model.Quota(name=params.name, description=params.description, amount=create_amount, operation=params.operation) self.sa_session.add(quota) # If this is a default quota, create the DefaultQuotaAssociation if params.default != 'no': self.app.quota_agent.set_default_quota(params.default, quota) message = "Default quota '%s' has been created." % quota.name else: # Create the UserQuotaAssociations in_users = [self.sa_session.query(self.app.model.User).get(decode_id(x) if decode_id else x) for x in util.listify(params.in_users)] in_groups = [self.sa_session.query(self.app.model.Group).get(decode_id(x) if decode_id else x) for x in util.listify(params.in_groups)] if None in in_users: raise ActionInputError("One or more invalid user id has been provided.") for user in in_users: uqa = self.app.model.UserQuotaAssociation(user, quota) self.sa_session.add(uqa) # Create the GroupQuotaAssociations if None in in_groups: raise ActionInputError("One or more invalid group id has been provided.") for group in in_groups: gqa = self.app.model.GroupQuotaAssociation(group, quota) self.sa_session.add(gqa) message = "Quota '%s' has been created with %d associated users and %d associated groups." % (quota.name, len(in_users), len(in_groups)) self.sa_session.flush() return quota, message def _rename_quota(self, quota, params): if not params.name: raise ActionInputError('Enter a valid name.') elif params.name != quota.name and self.sa_session.query(self.app.model.Quota).filter(self.app.model.Quota.table.c.name == params.name).first(): raise ActionInputError('A quota with that name already exists.') else: old_name = quota.name quota.name = params.name quota.description = params.description self.sa_session.add(quota) self.sa_session.flush() message = "Quota '%s' has been renamed to '%s'." % (old_name, params.name) return message def _manage_users_and_groups_for_quota(self, quota, params, decode_id=None): if quota.default: raise ActionInputError('Default quotas cannot be associated with specific users and groups.') else: in_users = [self.sa_session.query(self.app.model.User).get(decode_id(x) if decode_id else x) for x in util.listify(params.in_users)] if None in in_users: raise ActionInputError("One or more invalid user id has been provided.") in_groups = [self.sa_session.query(self.app.model.Group).get(decode_id(x) if decode_id else x) for x in util.listify(params.in_groups)] if None in in_groups: raise ActionInputError("One or more invalid group id has been provided.") self.app.quota_agent.set_entity_quota_associations(quotas=[quota], users=in_users, groups=in_groups) self.sa_session.refresh(quota) message = "Quota '%s' has been updated with %d associated users and %d associated groups." % (quota.name, len(in_users), len(in_groups)) return message def _edit_quota(self, quota, params): if params.amount.lower() in ('unlimited', 'none', 'no limit'): new_amount = None else: try: new_amount = util.size_to_bytes(params.amount) except AssertionError: new_amount = False if not params.amount: raise ActionInputError('Enter a valid amount.') elif new_amount is False: raise ActionInputError('Unable to parse the provided amount.') elif params.operation not in self.app.model.Quota.valid_operations: raise ActionInputError('Enter a valid operation.') else: quota.amount = new_amount quota.operation = params.operation self.sa_session.add(quota) self.sa_session.flush() message = "Quota '%s' is now '%s'." % (quota.name, quota.operation + quota.display_amount) return message def _set_quota_default(self, quota, params): if params.default != 'no' and params.default not in self.app.model.DefaultQuotaAssociation.types.__dict__.values(): raise ActionInputError('Enter a valid default type.') else: if params.default != 'no': self.app.quota_agent.set_default_quota(params.default, quota) message = "Quota '%s' is now the default for %s users." % (quota.name, params.default) else: if quota.default: message = "Quota '%s' is no longer the default for %s users." % (quota.name, quota.default[0].type) for dqa in quota.default: self.sa_session.delete(dqa) self.sa_session.flush() else: message = "Quota '%s' is not a default." % quota.name return message def _unset_quota_default(self, quota, params=None): if not quota.default: raise ActionInputError("Quota '%s' is not a default." % quota.name) else: message = "Quota '%s' is no longer the default for %s users." % (quota.name, quota.default[0].type) for dqa in quota.default: self.sa_session.delete(dqa) self.sa_session.flush() return message def _delete_quota(self, quota, params=None): quotas = util.listify(quota) names = [] for q in quotas: if q.default: names.append(q.name) if len(names) == 1: raise ActionInputError("Quota '%s' is a default, please unset it as a default before deleting it." % (names[0])) elif len(names) > 1: raise ActionInputError("Quotas are defaults, please unset them as defaults before deleting them: " + ', '.join(names)) message = "Deleted %d quotas: " % len(quotas) for q in quotas: q.deleted = True self.sa_session.add(q) names.append(q.name) self.sa_session.flush() message += ', '.join(names) return message def _undelete_quota(self, quota, params=None): quotas = util.listify(quota) names = [] for q in quotas: if not q.deleted: names.append(q.name) if len(names) == 1: raise ActionInputError("Quota '%s' has not been deleted, so it cannot be undeleted." % (names[0])) elif len(names) > 1: raise ActionInputError("Quotas have not been deleted so they cannot be undeleted: " + ', '.join(names)) message = "Undeleted %d quotas: " % len(quotas) for q in quotas: q.deleted = False self.sa_session.add(q) names.append(q.name) self.sa_session.flush() message += ', '.join(names) return message def _purge_quota(self, quota, params=None): """ This method should only be called for a Quota that has previously been deleted. Purging a deleted Quota deletes all of the following from the database: - UserQuotaAssociations where quota_id == Quota.id - GroupQuotaAssociations where quota_id == Quota.id """ quotas = util.listify(quota) names = [] for q in quotas: if not q.deleted: names.append(q.name) if len(names) == 1: raise ActionInputError("Quota '%s' has not been deleted, so it cannot be purged." % (names[0])) elif len(names) > 1: raise ActionInputError("Quotas have not been deleted so they cannot be undeleted: " + ', '.join(names)) message = "Purged %d quotas: " % len(quotas) for q in quotas: # Delete UserQuotaAssociations for uqa in q.users: self.sa_session.delete(uqa) # Delete GroupQuotaAssociations for gqa in q.groups: self.sa_session.delete(gqa) names.append(q.name) self.sa_session.flush() message += ', '.join(names) return message