Source code for q2_sdk.hq.db.tecton_config

import json
import uuid
from argparse import _SubParsersAction
from functools import partial
from typing import Optional

from q2_sdk.hq.models.online_session import OnlineSession
from q2_sdk.hq.models.online_user import OnlineUser
from q2_sdk.tools import utils
from .db_object import DbObject
from .ui_config_property_data import UiConfigPropertyData

CONFIG_PROPERTY_NAME = "Q2_GUI_TABLET_APPLICATION_SYSTEM_VARIABLE_TECTON"




class TectonConfig(DbObject):


    def add_arguments(self, parser: _SubParsersAction):
        subparser = parser.add_parser("get_tecton_config")
        subparser.set_defaults(parser="get_tecton_config")
        subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))




    async def get(self, serialize_for_cli=False, return_raw=False):
        ucpd_obj = UiConfigPropertyData(self.logger, hq_credentials=self.hq_credentials)
        row = await ucpd_obj.get(CONFIG_PROPERTY_NAME)
        json_row = {}
        if row and row[0].findtext("PropertyValue"):
            json_row = json.loads(row[0].PropertyValue.text)
        elif return_raw:
            return None
        response = json_row
        if serialize_for_cli:
            response = json.dumps(json_row, indent=2)
        return response




    async def create(self, feature_name: str, feature_manifest: dict) -> dict:
        return await self._create_or_update(feature_name, feature_manifest)




    async def update(self, feature_name: str, feature_manifest: dict) -> dict:
        return await self._create_or_update(feature_name, feature_manifest)




    @staticmethod
    def get_tecton_config(extension: str, short_name: str) -> Optional[dict]:
        frontend_dict = utils.get_feature_manifest(extension)

        if frontend_dict:
            for key in frontend_dict["modules"].keys():
                frontend_dict["modules"][key]["meta"].update({
                    "moduleName": key,
                    "featureName": short_name,
                })

        return frontend_dict




    async def delete(self, feature_name: str) -> bool:
        """Returns True if successful"""
        config = await self.get()
        if config is not None:
            features = config.get("features", {})
            features.pop(feature_name, None)
            config.update({"features": features})

            await self._update_config(config)
        return True


    async def _create_config(self, config: dict):
        ucpd_obj = UiConfigPropertyData(self.logger, hq_credentials=self.hq_credentials)
        await ucpd_obj.create(CONFIG_PROPERTY_NAME, "string", json.dumps(config))

    async def _update_config(self, config: dict):
        session = OnlineSession()
        session.session_id = str(uuid.uuid4())
        session.workstation = str(uuid.uuid4())
        online_user = OnlineUser()
        online_user.customer_id = 0
        online_user.user_id = 0
        online_user.user_logon_id = 0
        ucpd_obj = UiConfigPropertyData(self.logger, hq_credentials=self.hq_credentials)
        await ucpd_obj.update(
            CONFIG_PROPERTY_NAME,
            json.dumps(config),
            session,
            online_user,
            ui_source="OnlineBanking",
        )

    async def _create_or_update(
        self, feature_name: str, feature_manifest: dict
    ) -> dict:
        config = await self.get(return_raw=True)

        create = False
        if config is None:
            create = True
            config = {}

        features = config.get("features", {})

        if feature_name in features:
            features.pop(feature_name)

        features.update({feature_name: feature_manifest})
        config.update({"features": features})

        if create:
            await self._create_config(config)
        else:
            await self._update_config(config)

        return config



    async def upsert(self, config) -> dict:
        existing_config = await self.get(return_raw=True)
        create = existing_config is None

        config["features"] = config.get("features", {})

        if create:
            await self._create_config(config)
        else:
            await self._update_config(config)

        return config