Source code for luna_python_matcher.app_common.base_application

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

from typing import Generic

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 CONTEXT_CLASS, LunaApp
from crutches_on_wheels.cow.web.cmd_parser import ApplicationCMDParser
from crutches_on_wheels.cow.web.context import CONFIG_CLASS
from crutches_on_wheels.cow.web.request import VLRequest
from crutches_on_wheels.cow.web.utils import decodeBytes, isValidJsonRequestContentType, loadJson, loadMsgPack


[docs] class MatcherRequest(VLRequest): """ Matcher request """
[docs] 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
[docs] class MatcherBaseApp(LunaApp, Generic[CONFIG_CLASS, CONTEXT_CLASS]): """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