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