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
[docs]class SettingsHandler(BaseRequestHandler):
"""
Handler for creating or receiving settings.
"""
[docs] 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)
[docs] 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})
[docs] 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]})