Source code for q2_sdk.hq.db.q2_config_registry

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