import json
from typing import Optional
from q2_sdk.core.http_handlers.hq_handler import Q2HqRequestHandler
from q2_sdk.core.configuration import settings
[docs]
class Q2ArdentRequestHandler(Q2HqRequestHandler):
"""
RequestHandler meant to be used for requests incoming from Ardent.
"""
DESCRIPTION = ""
def __init__(self, application, request, **kwargs):
super().__init__(application, request, **kwargs)
self.allow_non_q2_traffic = True
self._form_fields = None
[docs]
async def prepare(self):
await super().prepare()
customer_key = self.request.headers.get("customerKey", "")
customer_key = customer_key.replace("CHANGEME", "")
if not customer_key:
wedge_json_data = json.loads(
self.request.headers.get("Wedgejsondata", "{}")
)
customer_key = wedge_json_data.get("customerKey", "")
if customer_key and customer_key != settings.VAULT_KEY:
supplied_hq = self._get_hq_from_key(customer_key)
if supplied_hq:
self.hq_credentials = supplied_hq
if self.hq_credentials:
await self.get_wedge_address_configs()
async def route_request(self, routing_key: Optional[str] = None):
if not routing_key:
routing_key = self.form_fields.get("routing_key")
if not routing_key:
routing_key = self.DEFAULT_ROUTE
if isinstance(routing_key, list):
if len(routing_key) > 1:
self.logger.error("Multiple routing keys present: '%s'.", routing_key)
routing_key = self.DEFAULT_ROUTE
else:
routing_key = routing_key[0].decode("utf-8")
if routing_key not in self.router:
self.logger.error(
"Routing key '%s' not present in router dictionary.", routing_key
)
routing_key = self.DEFAULT_ROUTE
route = self.router[routing_key]
self.logger.info("Transition: routing to '%s'", routing_key)
self.active_route = routing_key
return await self.call_route(route)
def parse_query_parameters(self) -> dict:
if not self.request.arguments:
return {}
return {
key: str(value[0], "utf-8") for key, value in self.request.arguments.items()
}
@property
def form_fields(self):
if not self._form_fields:
fields = self.parse_query_parameters()
if self.request.body:
overlay_fields = {}
try:
overlay_fields = json.loads(self.request.body)
except json.decoder.JSONDecodeError:
self.logger.error("Unable to decode non-JSON body")
fields.update(overlay_fields)
self._form_fields = fields
return self._form_fields