Source code for luna_tasks.app.handlers.base_handler
# -*- coding: utf-8 -*-
""" Base handler
Module realize base class for all handlers.
"""
from typing import Any
import marshmallow
from app.handlers.luna_tasks_functions import SubTaskSender
from common.handlers.base_handler import CommonBaseHandler
from crutches_on_wheels.errors.errors import Error
from crutches_on_wheels.errors.exception import VLException
[docs]class BaseRequestHandler(CommonBaseHandler):
"""
Base handler for other handlers.
"""
def __init__(self, request):
super().__init__(request)
self.luna3Client = self.app.ctx.luna3Session.getClient(self.requestId)
self.subtaskSender = SubTaskSender(
dbContext=self.dbContext,
logger=self.logger,
tasksWorkerAddress=self.config.lunaTasksWorkerAddress,
tasksWorkerTimeouts=self.config.lunaTasksWorkerTimeouts,
workerRequestsConcurrency=self.config.tasksRouterToWorkerRequestsConcurrency,
influxMonitoring=self.app.ctx.monitoring,
sendMonitoringData=self.config.monitoring.sendData,
)
[docs] async def loadMarshmallowFromJson(self, schema: marshmallow.Schema, returnLoadedData: bool = False) -> Any:
"""
Deserialize an input dictionary to an application-level data structure.
Args:
schema: marshmallow schema
returnLoadedData: whether to return loaded data from schema or not
Returns:
json as dict if validated
Raises:
VLException(Error.InvalidInputJson, 400, isCriticalError=False): if validation is failed
"""
try:
inputData = self.request.json
loadedData = schema.load(inputData)
if returnLoadedData:
return loadedData
return inputData
except marshmallow.ValidationError as e:
error = Error.InvalidInputJson.format(str(e))
raise VLException(error, 400, isCriticalError=False)