# -*- coding: utf-8 -*-
"""Limitation Handler
Module realize limitation handler.
"""
from typing import Optional
import ujson as json
from sanic.response import HTTPResponse
from app.handlers.base_handler import BaseRequestHandler
from app.handlers.schemas import NEW_LIMITATION_SCHEMA, PATCH_LIMITATION_SCHEMA
from crutches_on_wheels.errors.errors import Error
from crutches_on_wheels.errors.exception import VLException
[docs]class LimitationHandler(BaseRequestHandler):
"""
Handler for settings' limitation.
"""
async def _linkToGroups(self, limitationName: str, groupNames: Optional[str] = None):
"""
Link limitation to groups
Args:
limitationName: limitation names
groupNames: groups to link limitation to
Raises:
VLException(Error.formatError(Error.GroupsNotFound, nonExistGroups[0]), 400, False) if one or more groups
not found
"""
if groupNames is not None:
nonExistGroups = await self.dbContext.getNonExistGroups(groupNames)
if nonExistGroups:
raise VLException(Error.GroupsNotFound.format(nonExistGroups[0]), 400, False)
await self.dbContext.linkLimitationsToGroups(limitationList=[limitationName], groupList=groupNames)
[docs] async def put(self, limitationName: str) -> HTTPResponse:
"""
Request to put limitation, see `spec createLimitation`_
.. _`spec createLimitation`:
_static/api.html#operation/createLimitation
Resource is reached by address '/settings/limitations/{limitationName}'.
"""
limitationDict = self.request.json
self.validateJson(limitationDict, NEW_LIMITATION_SCHEMA, useJsonSchema=False)
self.validateJson(limitationDict["default_value"], limitationDict["validation_schema"], useJsonSchema=True)
await self.dbContext.deleteLimitation(limitationName=limitationName, ignoreExistence=True)
await self.dbContext.createOrReplaceLimitation(limitationName, limitationDict)
settingId = await self.dbContext.addSetting(
limitationName,
json.dumps(limitationDict["default_value"], ensure_ascii=False),
limitationDict.get("description"),
)
await self._linkToGroups(limitationName, limitationDict.get("group_names"))
self.respHeaders = {"Location": f"/{self.apiVersion}/settings/limitations/{limitationName}"}
return self.success(201, outputJson={"setting_id": settingId})
[docs] async def get(self, limitationName: str) -> HTTPResponse:
"""
Request to receive settings limitations, see `spec getLimitation`_
.. _`spec getLimitation`:
_static/api.html#operation/getLimitation
Resource is reached by address '/settings/limitations/{limitationName}'.
"""
response = await self.dbContext.getLimitation(limitationName)
return self.success(200, outputJson=response)
[docs] async def patch(self, limitationName: str) -> HTTPResponse:
"""
Request to patch limitation, see `spec patchLimitation`_
.. _`spec patchLimitation`:
_static/api.html#operation/patchLimitation
Resource is reached by address '/settings/limitations/{limitationName}'.
"""
limitationDict = self.request.json
self.validateJson(limitationDict, PATCH_LIMITATION_SCHEMA, useJsonSchema=False)
oldLimitation = await self.dbContext.getLimitation(limitationName=limitationName)
defaultValue = limitationDict.get("default_value", None)
if defaultValue is None:
defaultValue = oldLimitation["default_value"]
validationSchema = limitationDict.get("validation_schema") or oldLimitation["validation_schema"]
self.validateJson(defaultValue, validationSchema, useJsonSchema=True)
isNeedToValidateOldSettings = "validation_schema" in limitationDict
await self.dbContext.updateLimitation(limitationName, limitationDict)
oldDefaultSetting = (await self.dbContext.getSettingsByFilters(settingNames=[limitationName], isDefault=True))[
0
][0]
await self.dbContext.replaceSetting(
settingId=oldDefaultSetting["id"],
name=limitationName,
value=json.dumps(limitationDict["default_value"], ensure_ascii=False)
if "default_value" in limitationDict
else oldDefaultSetting["value"],
description=limitationDict["description"]
if "description" in limitationDict
else oldDefaultSetting["description"],
)
await self._linkToGroups(limitationName, limitationDict.get("group_names"))
unvalidatedSettingIds = []
if isNeedToValidateOldSettings:
oldSettings = (
await self.dbContext.getSettingsByFilters(settingNames=[limitationName], pageSize=-1, isDefault=False)
)[0]
for oldSetting in oldSettings:
try:
self.validateJson(oldSetting["value"], limitationDict["validation_schema"], useJsonSchema=True)
except VLException:
unvalidatedSettingIds.append(oldSetting["id"])
resJson = {"setting_id": oldDefaultSetting["id"], "unverified_setting_ids": unvalidatedSettingIds}
self.respHeaders = {"Location": f"/{self.apiVersion}/settings/limitations/{limitationName}"}
return self.success(200, outputJson=resJson)
[docs] async def delete(self, limitationName: str) -> HTTPResponse:
"""
Request to delete limitation, see `spec deleteLimitation`_
.. _`spec deleteLimitation`:
_static/api.html#operation/deleteLimitation
Resource is reached by address '/settings/limitations/{limitationName}'.
"""
await self.dbContext.deleteLimitation(limitationName)
return self.success(204)
