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)