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 galaxy.managers.quotas
"""
Manager and Serializers for Quotas.
For more information about quotas: https://galaxyproject.org/admin/disk-quotas/
"""
import logging
from typing import (
cast,
Optional,
Tuple,
Union,
)
from galaxy import model, util
from galaxy.app import StructuredApp
from galaxy.exceptions import ActionInputError
from galaxy.managers import base
from galaxy.quota import DatabaseQuotaAgent
from galaxy.quota._schema import (
CreateQuotaParams,
DefaultQuotaValues,
QuotaOperation,
)
from galaxy.schema.fields import EncodedDatabaseIdField
log = logging.getLogger(__name__)
[docs]class QuotaManager:
"""Interface/service object to interact with Quotas."""
@property
def sa_session(self):
return self.app.model.context
@property
def quota_agent(self) -> DatabaseQuotaAgent:
return cast(DatabaseQuotaAgent, self.app.quota_agent)
[docs] def create_quota(self, payload: dict, decode_id=None) -> Tuple[model.Quota, str]:
params = CreateQuotaParams.parse_obj(payload)
create_amount = self._parse_amount(params.amount)
if self.sa_session.query(model.Quota).filter(model.Quota.name == params.name).first():
raise ActionInputError("Quota names must be unique and a quota with that name already exists, please choose another name.")
elif create_amount is False:
raise ActionInputError("Unable to parse the provided amount.")
elif params.operation not in model.Quota.valid_operations:
raise ActionInputError("Enter a valid operation.")
elif params.default != DefaultQuotaValues.NO and params.operation != QuotaOperation.EXACT:
raise ActionInputError("Operation for a default quota must be '='.")
elif create_amount is None and params.operation != QuotaOperation.EXACT:
raise ActionInputError("Operation for an unlimited quota must be '='.")
# Create the quota
quota = 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 != DefaultQuotaValues.NO:
self.quota_agent.set_default_quota(params.default, quota)
message = f"Default quota '{quota.name}' has been created."
else:
# Create the UserQuotaAssociations
in_users = [self.sa_session.query(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(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 = 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 = model.GroupQuotaAssociation(group, quota)
self.sa_session.add(gqa)
message = f"Quota '{quota.name}' has been created with {len(in_users)} associated users and {len(in_groups)} associated groups."
self.sa_session.flush()
return quota, message
def _parse_amount(self, amount: str) -> Optional[Union[int, bool]]:
if amount.lower() in ('unlimited', 'none', 'no limit'):
return None
try:
return util.size_to_bytes(amount)
except AssertionError:
return False
[docs] def rename_quota(self, quota, params) -> str:
if not params.name:
raise ActionInputError('Enter a valid name.')
elif params.name != quota.name and self.sa_session.query(model.Quota).filter(model.Quota.name == params.name).first():
raise ActionInputError('A quota with that name already exists.')
else:
old_name = quota.name
quota.name = params.name
if params.description:
quota.description = params.description
self.sa_session.add(quota)
self.sa_session.flush()
message = f"Quota '{old_name}' has been renamed to '{params.name}'."
return message
[docs] def manage_users_and_groups_for_quota(self, quota, params, decode_id=None) -> str:
if quota.default:
raise ActionInputError('Default quotas cannot be associated with specific users and groups.')
else:
in_users = [self.sa_session.query(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(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.quota_agent.set_entity_quota_associations(quotas=[quota], users=in_users, groups=in_groups)
self.sa_session.refresh(quota)
message = f"Quota '{quota.name}' has been updated with {len(in_users)} associated users and {len(in_groups)} associated groups."
return message
[docs] def edit_quota(self, quota, params) -> str:
if params.amount.lower() in ('unlimited', 'none', 'no limit'):
new_amount = None
else:
try:
new_amount = util.size_to_bytes(params.amount)
except (AssertionError, ValueError):
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 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 = f"Quota '{quota.name}' is now '{quota.operation}{quota.display_amount}'."
return message
[docs] def set_quota_default(self, quota, params) -> str:
if params.default != 'no' and params.default not in model.DefaultQuotaAssociation.types.__members__.values():
raise ActionInputError('Enter a valid default type.')
else:
if params.default != 'no':
self.quota_agent.set_default_quota(params.default, quota)
message = f"Quota '{quota.name}' is now the default for {params.default} users."
else:
if quota.default:
message = f"Quota '{quota.name}' is no longer the default for {quota.default[0].type} users."
for dqa in quota.default:
self.sa_session.delete(dqa)
self.sa_session.flush()
else:
message = f"Quota '{quota.name}' is not a default."
return message
[docs] def unset_quota_default(self, quota, params=None) -> str:
if not quota.default:
raise ActionInputError(f"Quota '{quota.name}' is not a default.")
else:
message = f"Quota '{quota.name}' is no longer the default for {quota.default[0].type} users."
for dqa in quota.default:
self.sa_session.delete(dqa)
self.sa_session.flush()
return message
[docs] def delete_quota(self, quota, params=None) -> str:
quotas = util.listify(quota)
names = []
for q in quotas:
if q.default:
names.append(q.name)
if len(names) == 1:
raise ActionInputError(f"Quota '{names[0]}' is a default, please unset it as a default before deleting it.")
elif len(names) > 1:
raise ActionInputError(f"Quotas are defaults, please unset them as defaults before deleting them: {', '.join(names)}")
message = f"Deleted {len(quotas)} 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
[docs] def undelete_quota(self, quota, params=None) -> str:
quotas = util.listify(quota)
names = []
for q in quotas:
if not q.deleted:
names.append(q.name)
if len(names) == 1:
raise ActionInputError(f"Quota '{names[0]}' has not been deleted, so it cannot be undeleted.")
elif len(names) > 1:
raise ActionInputError(f"Quotas have not been deleted so they cannot be undeleted: {', '.join(names)}")
message = f"Undeleted {len(quotas)} 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
[docs] 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(f"Quota '{names[0]}' has not been deleted, so it cannot be purged.")
elif len(names) > 1:
raise ActionInputError(f"Quotas have not been deleted so they cannot be undeleted: {', '.join(names)}")
message = f"Purged {len(quotas)} 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
[docs] def get_quota(self, trans, id: EncodedDatabaseIdField, deleted: Optional[bool] = None) -> model.Quota:
return base.get_object(trans, id, 'Quota', check_ownership=False, check_accessible=False, deleted=deleted)