from argparse import _SubParsersAction
from dataclasses import dataclass, fields
from enum import Enum
from functools import partial
from lxml import objectify
from requests import RequestException
from q2_sdk.core import q2_requests
from q2_sdk.core.configuration import settings
from q2_sdk.hq.table_row import TableRow
from .db_object import DbObject
[docs]
@dataclass
class Q2ConfigParameters:
extensionId: int = None
featureName: str = None
displayName: str = None
featureFlag: str = None
configUrl: str = None
[docs]
def build_json_parameters(self):
json_param = {}
for field in fields(self):
name = field.name
value = getattr(self, name)
if value:
json_param[name] = value
return json_param
[docs]
class Q2ConfigTable(Enum):
extension_id = "ExtensionId"
feature_name = "FeatureName"
display_name = "DisplayName"
feature_flag = "FeatureFlag"
config_url = "ConfigUrl"
[docs]
class Q2ConfigRegistryRow(TableRow):
ExtensionId: int
FeatureName: str
DisplayName: str
FeatureFlag: str
ConfigUrl: str
[docs]
class Q2ConfigRegistry(DbObject):
# GET_BY_NAME_KEY = "column in the db response"
NAME = "Q2ConfigRegistry"
REPRESENTATION_ROW_CLASS = Q2ConfigRegistryRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_q2_config_registry_entries")
subparser.set_defaults(parser="get")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser = parser.add_parser("get_q2_config_registry_entry")
subparser.set_defaults(parser="get_by_extension_id")
subparser.set_defaults(
func=partial(self.get_by_extension_id, serialize_for_cli=True)
)
subparser.add_argument(
"-e",
"--extension-id",
help="q2_config.public.extensions_registry.extension_id",
)
[docs]
def serialize(self, data):
columns = [column.name for column in Q2ConfigTable]
return self.serialize_for_cli(data, columns)
[docs]
def to_xml(self, json_data):
xml_dict = []
for data in json_data:
root = objectify.Element("Table")
for key, value in data.items():
objectify.SubElement(root, Q2ConfigTable[key].value)._setText(
str(value)
)
xml_dict.append(root)
return xml_dict
[docs]
async def get(self, serialize_for_cli=False) -> list[Q2ConfigRegistryRow]:
data = []
try:
response = await q2_requests.get(
self.logger, settings.Q2CONFIG_REGISTRY_API_URL
)
if response.ok:
data = response.json()["data"]
except RequestException:
data = []
if self.ret_table_obj:
xml = self.to_xml(data)
data = [Q2ConfigRegistryRow(row) for row in xml]
if serialize_for_cli:
data = self.serialize(data)
return data
[docs]
async def get_by_extension_id(self, extension_id) -> Q2ConfigRegistryRow:
data = None
response = await q2_requests.get(
self.logger,
f"{settings.Q2CONFIG_REGISTRY_API_URL}?extensionId={extension_id}",
)
response.raise_for_status()
data = response.json()["data"]
if self.ret_table_obj:
xml = self.to_xml(data)
data = [Q2ConfigRegistryRow(row) for row in xml]
return data[0]
[docs]
async def create(self, endpoint_parameters: Q2ConfigParameters):
json_parameters = endpoint_parameters.build_json_parameters()
response = await q2_requests.post(
self.logger, settings.Q2CONFIG_REGISTRY_API_URL, json=json_parameters
)
response.raise_for_status()
return response
[docs]
async def update(self, endpoint_parameters: Q2ConfigParameters):
json_parameters = endpoint_parameters.build_json_parameters()
response = await q2_requests.post(
self.logger, settings.Q2CONFIG_REGISTRY_API_URL, json=json_parameters
)
response.raise_for_status()
return response
[docs]
async def delete(self, extension_id):
response = await q2_requests.put(
self.logger,
settings.Q2CONFIG_REGISTRY_DELETE_API_URL,
json={"extensionId": extension_id},
)
response.raise_for_status()
return response