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.quota
"""
Galaxy Quotas
"""
import logging
import galaxy.util
log = logging.getLogger(__name__)
[docs]class NoQuotaAgent:
"""Base quota agent, always returns no quota"""
@property
def default_quota(self):
return None
[docs] def get_usage(self, trans=None, user=False, history=False):
if trans:
user = trans.user
history = trans.history
assert user is not False, "Could not determine user."
if not user:
assert history, "Could not determine anonymous user's history."
usage = history.disk_size
else:
usage = user.total_disk_usage
return usage
[docs] def get_percent(self, trans=None, user=False, history=False, usage=False, quota=False):
return None
[docs]class QuotaAgent(NoQuotaAgent):
"""Class that handles galaxy quotas"""
[docs] def get_quota(self, user, nice_size=False):
"""
Calculated like so:
1. Anonymous users get the default quota.
2. Logged in users start with the highest of their associated '='
quotas or the default quota, if there are no associated '='
quotas. If an '=' unlimited (-1 in the database) quota is found
during this process, the user has no quota (aka unlimited).
3. Quota is increased or decreased by any corresponding '+' or '-'
quotas.
"""
if not user:
return self.default_unregistered_quota
quotas = []
for group in [uga.group for uga in user.groups]:
for quota in [gqa.quota for gqa in group.quotas]:
if quota not in quotas:
quotas.append(quota)
for quota in [uqa.quota for uqa in user.quotas]:
if quota not in quotas:
quotas.append(quota)
use_default = True
max = 0
adjustment = 0
rval = 0
for quota in quotas:
if quota.deleted:
continue
if quota.operation == '=' and quota.bytes == -1:
rval = None
break
elif quota.operation == '=':
use_default = False
if quota.bytes > max:
max = quota.bytes
elif quota.operation == '+':
adjustment += quota.bytes
elif quota.operation == '-':
adjustment -= quota.bytes
if use_default:
max = self.default_registered_quota
if max is None:
rval = None
if rval is not None:
rval = max + adjustment
if rval <= 0:
rval = 0
if nice_size:
if rval is not None:
rval = galaxy.util.nice_size(rval)
else:
rval = 'unlimited'
return rval
@property
def default_unregistered_quota(self):
return self._default_quota(self.model.DefaultQuotaAssociation.types.UNREGISTERED)
@property
def default_registered_quota(self):
return self._default_quota(self.model.DefaultQuotaAssociation.types.REGISTERED)
def _default_quota(self, default_type):
dqa = self.sa_session.query(self.model.DefaultQuotaAssociation).filter(self.model.DefaultQuotaAssociation.table.c.type == default_type).first()
if not dqa:
return None
if dqa.quota.bytes < 0:
return None
return dqa.quota.bytes
[docs] def set_default_quota(self, default_type, quota):
# Unset the current default(s) associated with this quota, if there are any
for dqa in quota.default:
self.sa_session.delete(dqa)
# Unset the current users/groups associated with this quota
for uqa in quota.users:
self.sa_session.delete(uqa)
for gqa in quota.groups:
self.sa_session.delete(gqa)
# Find the old default, assign the new quota if it exists
dqa = self.sa_session.query(self.model.DefaultQuotaAssociation).filter(self.model.DefaultQuotaAssociation.table.c.type == default_type).first()
if dqa:
dqa.quota = quota
# Or create if necessary
else:
dqa = self.model.DefaultQuotaAssociation(default_type, quota)
self.sa_session.add(dqa)
self.sa_session.flush()
[docs] def get_percent(self, trans=None, user=False, history=False, usage=False, quota=False):
"""
Return the percentage of any storage quota applicable to the user/transaction.
"""
# if trans passed, use it to get the user, history (instead of/override vals passed)
if trans:
user = trans.user
history = trans.history
# if quota wasn't passed, attempt to get the quota
if quota is False:
quota = self.get_quota(user)
# return none if no applicable quotas or quotas disabled
if quota is None:
return None
# get the usage, if it wasn't passed
if usage is False:
usage = self.get_usage(trans, user, history)
try:
return min((int(float(usage) / quota * 100), 100))
except ZeroDivisionError:
return 100
[docs] def set_entity_quota_associations(self, quotas=None, users=None, groups=None, delete_existing_assocs=True):
if quotas is None:
quotas = []
if users is None:
users = []
if groups is None:
groups = []
for quota in quotas:
if delete_existing_assocs:
flush_needed = False
for a in quota.users + quota.groups:
self.sa_session.delete(a)
flush_needed = True
if flush_needed:
self.sa_session.flush()
for user in users:
uqa = self.model.UserQuotaAssociation(user, quota)
self.sa_session.add(uqa)
for group in groups:
gqa = self.model.GroupQuotaAssociation(group, quota)
self.sa_session.add(gqa)
self.sa_session.flush()
[docs] def get_user_quotas(self, user):
rval = []
if not user:
dqa = self.sa_session.query(self.model.DefaultQuotaAssociation) \
.filter(self.model.DefaultQuotaAssociation.table.c.type == self.model.DefaultQuotaAssociation.types.UNREGISTERED).first()
if dqa:
rval.append(dqa.quota)
else:
dqa = self.sa_session.query(self.model.DefaultQuotaAssociation) \
.filter(self.model.DefaultQuotaAssociation.table.c.type == self.model.DefaultQuotaAssociation.types.REGISTERED).first()
if dqa:
rval.append(dqa.quota)
for uqa in user.quotas:
rval.append(uqa.quota)
for group in [uga.group for uga in user.groups]:
for gqa in group.quotas:
rval.append(gqa.quota)
return rval