Source code for galaxy.webapps.galaxy.services.quotas

import logging
from typing import Optional

from sqlalchemy import (
    false,
    select,
    true,
)
from sqlalchemy.orm import Session

from galaxy import util
from galaxy.managers.context import ProvidesUserContext
from galaxy.managers.groups import get_group_by_name
from galaxy.managers.quotas import QuotaManager
from galaxy.managers.users import get_user_by_email
from galaxy.model import Quota
from galaxy.quota._schema import (
    CreateQuotaParams,
    CreateQuotaResult,
    DefaultQuotaValues,
    DeleteQuotaPayload,
    QuotaDetails,
    QuotaSummaryList,
    UpdateQuotaParams,
)
from galaxy.schema.fields import DecodedDatabaseIdField
from galaxy.security.idencoding import IdEncodingHelper
from galaxy.web import url_for
from galaxy.webapps.galaxy.services.base import ServiceBase

log = logging.getLogger(__name__)


[docs]class QuotasService(ServiceBase):
    """Interface/service object shared by controllers for interacting with quotas."""

[docs]    def __init__(self, security: IdEncodingHelper, quota_manager: QuotaManager):
        super().__init__(security)
        self.quota_manager = quota_manager

[docs]    def index(self, trans: ProvidesUserContext, deleted: bool = False) -> QuotaSummaryList:
        """Displays a list of quotas."""
        rval = []
        if deleted:
            route = "deleted_quota"
            quotas = get_quotas(trans.sa_session, deleted=True)
        else:
            route = "quota"
            quotas = get_quotas(trans.sa_session, deleted=False)
        for quota in quotas:
            item = quota.to_dict(value_mapper={"id": DecodedDatabaseIdField.encode})
            encoded_id = DecodedDatabaseIdField.encode(quota.id)
            item["url"] = url_for(route, id=encoded_id)
            rval.append(item)
        return QuotaSummaryList.construct(__root__=rval)

[docs]    def show(self, trans: ProvidesUserContext, id: DecodedDatabaseIdField, deleted: bool = False) -> QuotaDetails:
        """Displays information about a quota."""
        quota = self.quota_manager.get_quota(trans, id, deleted=deleted)
        rval = quota.to_dict(
            view="element", value_mapper={"id": DecodedDatabaseIdField.encode, "total_disk_usage": float}
        )
        return QuotaDetails.construct(**rval)

[docs]    def create(self, trans: ProvidesUserContext, params: CreateQuotaParams) -> CreateQuotaResult:
        """Creates a new quota."""
        payload = params.dict()
        self.validate_in_users_and_groups(trans, payload)
        quota, message = self.quota_manager.create_quota(payload)
        item = quota.to_dict(value_mapper={"id": DecodedDatabaseIdField.encode})
        item["url"] = url_for("quota", id=DecodedDatabaseIdField.encode(quota.id))
        item["message"] = message
        return CreateQuotaResult.construct(**item)

[docs]    def update(self, trans: ProvidesUserContext, id: DecodedDatabaseIdField, params: UpdateQuotaParams) -> str:
        """Modifies a quota."""
        payload = params.dict()
        self.validate_in_users_and_groups(trans, payload)
        quota = self.quota_manager.get_quota(trans, id, deleted=False)

        params = UpdateQuotaParams(**payload)
        # FIXME: Doing it this way makes the update non-atomic if a method fails after an earlier one has succeeded.
        methods = []
        if params.name or params.description:
            methods.append(self.quota_manager.rename_quota)
        if params.amount:
            methods.append(self.quota_manager.edit_quota)
        if params.default == DefaultQuotaValues.NO:
            methods.append(self.quota_manager.unset_quota_default)
        elif params.default:
            methods.append(self.quota_manager.set_quota_default)
        if params.in_users or params.in_groups:
            methods.append(self.quota_manager.manage_users_and_groups_for_quota)

        messages = []
        for method in methods:
            message = method(quota, params)
            messages.append(message)
        return "; ".join(messages)

[docs]    def delete(
        self, trans: ProvidesUserContext, id: DecodedDatabaseIdField, payload: Optional[DeleteQuotaPayload] = None
    ) -> str:
        """Marks a quota as deleted."""
        quota = self.quota_manager.get_quota(
            trans, id, deleted=False
        )  # deleted quotas are not technically members of this collection
        message = self.quota_manager.delete_quota(quota)
        if payload and payload.purge:
            message += self.quota_manager.purge_quota(quota)
        return message

[docs]    def undelete(self, trans: ProvidesUserContext, id: DecodedDatabaseIdField) -> str:
        """Restores a previously deleted quota."""
        quota = self.quota_manager.get_quota(trans, id, deleted=True)
        return self.quota_manager.undelete_quota(quota)

[docs]    def validate_in_users_and_groups(self, trans, payload):
        """
        For convenience, in_users and in_groups can be encoded IDs or emails/group names in the API.
        """

        def get_user_id(item):
            try:
                return trans.security.decode_id(item)
            except Exception:
                return get_user_by_email(trans.sa_session, item).id

        def get_group_id(item):
            try:
                return trans.security.decode_id(item)
            except Exception:
                return get_group_by_name(trans.sa_session, item).id

        new_in_users = []
        new_in_groups = []
        invalid = []
        for item in util.listify(payload.get("in_users", [])):
            try:
                new_in_users.append(get_user_id(item))
            except Exception:
                invalid.append(item)
        for item in util.listify(payload.get("in_groups", [])):
            try:
                new_in_groups.append(get_group_id(item))
            except Exception:
                invalid.append(item)
        if invalid:
            msg = (
                f"The following value(s) for associated users and/or groups could not be parsed: {', '.join(invalid)}."
            )
            msg += "  Valid values are email addresses of users, names of groups, or IDs of both."
            raise Exception(msg)
        payload["in_users"] = list(map(str, new_in_users))
        payload["in_groups"] = list(map(str, new_in_groups))


[docs]def get_quotas(session: Session, deleted: bool = False):
    is_deleted = true()
    if not deleted:
        is_deleted = false()
    stmt = select(Quota).where(Quota.deleted == is_deleted)
    return session.scalars(stmt)