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.cow.errors.errors import Error
from crutches_on_wheels.cow.errors.exception import VLException


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,
        )

    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)