from abc import ABC, abstractmethod
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)
def __str__(self):
def parse_params(item):
key, value = item
if "password" in key.lower():
value = "****"
return f"{key}={value}"
params = ", ".join(map(parse_params, self.request_params.items()))
return f"{self.__class__.__qualname__}({params})"
[docs]
class BaseParameter(ABC):
@abstractmethod
def serialize_as_xml(self): ...
@abstractmethod
def serialize_as_json(self) -> dict | list: ...
def __str__(self):
serialized = self.serialize_as_json()
match serialized:
case dict():
params = ", ".join(
f"{key}={value}" for key, value in serialized.items()
)
case list():
params = ", ".join(str(item) for item in serialized)
return f"{self.__class__.__qualname__}({params})"