from typing import Optional
from q2_sdk.core.q2_logging.logger import Q2LoggerType
from q2_sdk.hq.models.hq_credentials import HqCredentials
from q2_sdk.core.configuration import settings
from q2_sdk.tools import utils
from q2_sdk.core import contexts
[docs]
class BaseParamsObj:
"""
All HQ Endpoints will define an execute method which takes in an instance of BaseParamsObj
"""
ENDPOINT_PATH = None
def __init__(
self,
logger: Q2LoggerType,
hq_credentials: Optional[HqCredentials] = None,
):
"""
Builds the hq_url based on the defined ENDPOINT_PATH
"""
if not hq_credentials:
current_request = contexts.get_current_request(raise_if_none=False)
if current_request:
hq_credentials = current_request.request_handler.hq_credentials
else:
hq_credentials = settings.HQ_CREDENTIALS
self.hq_credentials: Optional[HqCredentials] = hq_credentials
self.logger = logger
self.hq_url = "{}/{}".format(self.hq_credentials.hq_url, self.ENDPOINT_PATH)
self.request_params = {}
def __getattr__(self, attr):
pascalized = utils.pascalize(attr)
if pascalized in self.request_params:
return self.request_params[pascalized]
return super().__getattribute__(attr)
[docs]
class BaseParameter:
def serialize_as_xml(self):
raise NotImplementedError
def serialize_as_json(self):
raise NotImplementedError