Source code for luna_python_matcher.app_common.base_application

"""
Module contains base application and request class.
"""

from vlutils.helpers import bytesToBase64

from app_common.global_vars.api_version import API_VERSION
from configs.config_common import RESPONSE_TIMEOUT
from crutches_on_wheels.cow.errors.errors import Error
from crutches_on_wheels.cow.errors.exception import VLException
from crutches_on_wheels.cow.web.application import LunaApp
from crutches_on_wheels.cow.web.cmd_parser import ApplicationCMDParser
from crutches_on_wheels.cow.web.request import VLRequest
from crutches_on_wheels.cow.web.utils import decodeBytes, isValidJsonRequestContentType, loadJson, loadMsgPack




class MatcherRequest(VLRequest):
    """
    Matcher request
    """



    def loadData(self, convertBytes: bool = False) -> dict:
        """
        Loads request data from json / msgpack.
        Args:
            convertBytes: whether to load data from msgpack with a custom hook to convert bytes
                          making data suitable for jsonschema-rs validation
        """

        if self.parsed_json is None:
            try:
                if self.content_type == "application/msgpack":
                    self.parsed_json = loadMsgPack(
                        self.body,
                        object_hook=self.decodeObjectBytes if convertBytes else None,
                    )
                elif isValidJsonRequestContentType(self.content_type):
                    text = decodeBytes(body=self.body, encoding=self.charset)
                    self.parsed_json = loadJson(text)
                else:
                    raise VLException(Error.BadContentType, 400, isCriticalError=False)
                return self.parsed_json
            except ValueError as exc:
                raise VLException(Error.RequestNotContainsJson, 400, isCriticalError=False) from exc
        return self.parsed_json


    @staticmethod
    def decodeObjectBytes(obj):
        for key in obj:
            if isinstance(obj[key], bytes):
                obj[key] = bytesToBase64(obj[key])
        return obj





class MatcherBaseApp[TConfig: BaseServiceSettingsClass, TCtx: LunaAppCtx](LunaApp):
    """Base service application"""

    apiVersion = API_VERSION
    requestCls = MatcherRequest

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.sanic.config["RESPONSE_TIMEOUT"] = RESPONSE_TIMEOUT

    def _getCMDArgParser(self) -> ApplicationCMDParser:
        """
        Get command line arguments parser. Overloading `_getCMDArgParser` to get the service type.
        Returns:
            parser with options
        """
        argParser = super()._getCMDArgParser()
        argParser.argParser.add_argument(
            "--service-type", type=str, choices=["matcher", "proxy"], help='service type: one of "matcher" or "proxy"'
        )
        return argParser