Source code for luna_configurator.app.handlers.settings_handler

# -*- coding: utf-8 -*-
"""Settings Handler

Module realize settings handler.
"""
from typing import Callable

import fastjsonschema
from sanic.response import HTTPResponse

from app.handlers.base_handler import BaseRequestHandler
from app.version import VERSION
from crutches_on_wheels.errors.errors import Error
from crutches_on_wheels.errors.exception import VLException
from crutches_on_wheels.web.query_getters import int01Getter, listStringsGetter
from db.functions import inputSetting


class SettingsHandler(BaseRequestHandler):
    """
    Handler for creating or receiving settings.
    """

    async def getNewSettingSchema(self, settingName: str) -> Callable[[dict], None]:
        """
        Get validation schema for new setting
        Args:
            settingName: setting name
        Returns:
            validation schema for new setting
        """
        validationSchema = await self.getSettingSchema(settingName)
        validationSchema["properties"]["tags"]["minItems"] = 1
        return fastjsonschema.compile(validationSchema)

    async def post(self) -> HTTPResponse:
        """
        Request to receive settings. See `spec newSetting`_.

        .. _`spec newSetting`:
            _static/api.html#operation/newSetting

        """
        settingDict = self.request.json
        settingName = settingDict.get("name", None)
        if not isinstance(settingName, str):
            raise VLException(Error.BadInputJson, 400, False)
        else:
            if settingName in (await self.dbContext.getAllSettingNames()):
                self.validateJson(settingDict, (await self.getNewSettingSchema(settingName)), useJsonSchema=False)
            else:
                raise VLException(Error.LimitationNotFound.format(settingName), 400, False)

        setting = inputSetting(settingDict)

        settingId = await self.dbContext.addSetting(**setting)
        self.respHeaders = {"Location": "/{}/settings/{}".format(VERSION["Version"]["api"], settingId)}
        return self.success(201, outputJson={"setting_id": settingId})

    async def get(self) -> HTTPResponse:
        """
        Request to receive settings according to provided filters. See `spec getSettings`_.

        .. _`spec getSettings`:
            _static/api.html#operation/getSettings

        """
        page, pageSize = self.getPagination()
        tags = self.getQueryParam("tags", lambda x: x.split(",") if x else None)
        description = self.getQueryParam("description")
        serviceName = self.getQueryParam("service_name")
        settingNames = [self.getQueryParam("setting_name")]
        getSchema = bool(self.getQueryParam("get_schema", int01Getter, default=False, require=False))
        isDefault = self.getQueryParam(
            "is_default", lambda value: bool(int01Getter(value)), default=None, require=False
        )
        groupNames = self.getQueryParam("group_names", listStringsGetter)
        if serviceName is not None and serviceName not in (await self.dbContext.getAllServiceNames()):
            raise VLException(Error.ServiceNameNotFound.format(serviceName), 400, False)

        if settingNames == [None]:
            settingNames = [
                limitation["limitation_name"]
                for limitation in (await self.dbContext.getLimitations())
                if serviceName in limitation["services"]
            ]

        response = await self.dbContext.getSettingsByFilters(
            settingNames,
            tags=tags,
            description=description,
            groupNames=groupNames,
            page=page,
            pageSize=pageSize,
            getSchema=getSchema,
            isDefault=isDefault,
        )
        return self.success(200, outputJson={"settings": response[0], "count": response[1]})