import asyncio
import json
from typing import Type, Optional
from q2_sdk.core.cli import textui
from q2_sdk.core.exceptions import DatabaseDataError, DevOnlyError
from q2_sdk.core.q2_logging.logger import Q2LoggerType
from q2_sdk.hq.db.form import Form
from q2_sdk.hq.models.hq_credentials import HqCredentials
from q2_sdk.hq.db.ui_text import UiText
from q2_sdk.hq.models.ui_text import UiText as UiTextModel
from q2_sdk.models.db_plan_shapes import UserPropertyFeature
from q2_sdk.models.installers.form_params import DbPlanRequirements
from q2_sdk.models.recursive_encoder import RecursiveEncoder
from q2_sdk.models.installers.form_params import SendAccountDetailsOption
from .admin_user_property_data_element import AdminUserPropertyDataElement
from .api_stored_proc import ApiStoredProc
from .audit_action import AuditAction
from .audit_category import AuditCategory
from .base import InstallStep, InstallStepAttribute
from .data_feed import DataFeed
from .disclaimer import Disclaimer
from .gt_data_element import GTDataElement
from .gt_flavor import GtFlavor
from .host_account_data_element import HostAccountDataElement
from .host_tran_code import HostTranCode
from .host_tran_code_group import HostTranCodeGroup
from .product import Product
from .product_type import ProductType
from .report_info import ReportInfo
from .required_form import RequiredForm
from .tecton_config import TectonConfig
from .third_party_data_element import ThirdPartyDataElement
from .ui_config_property_data import UIConfigPropertyData
from .ui_text_element import UiTextElement
from .uux_payload import UUXPayload
from .user_account_attribute import UserAccountAttribute
from .user_property_data import UserPropertyData
from .user_property_data_element import UserPropertyDataElement
from .vault_info import VaultInfo
from .vendor import Vendor
from .vendor_address import VendorAddress
from .vendor_config import VendorConfig
SCALAR_KEYS = (
"allow_prompts",
"disallow_add_to_nav",
"send_account_list",
"send_account_details",
"account_rights_bit_flag",
"payload_stored_proc",
"ui_text_prefix",
"insight_features",
"marketplace_features",
"found_unknown_db_plan_attrs",
"upde_allow_user_edit",
"upde_allow_user_view",
"upde_allow_cust_view",
)
def create_install_step_attr(data: dict) -> InstallStepAttribute:
return InstallStepAttribute(**{
key: value for key, value in data.items() if key != "default"
})
def get_install_step(db_plan: "DbPlan", attr_name: str) -> InstallStep:
if not hasattr(db_plan, attr_name):
return InstallStep()
else:
return getattr(db_plan, attr_name)[0]
async def force_ui_text_updates(logger, hq_credentials, ui_text_objs):
new_ui_text_objs = []
update_ui_text_list = []
ui_text_obj = UiText(logger, hq_credentials, ret_table_obj=True)
ui_language = "USEnglish"
for obj in ui_text_objs:
ui_text = await ui_text_obj.get(obj.prefixed_short_name)
if len(ui_text) > 0:
if obj.language:
ui_language = obj.language
update_ui_text_list.append(
ui_text_obj.update(obj.prefixed_short_name, obj.text_value, ui_language)
)
else:
new_ui_text_objs.append(obj)
if new_ui_text_objs:
await UiText(logger, hq_credentials).create_bulk(new_ui_text_objs)
if update_ui_text_list:
await asyncio.gather(*update_ui_text_list) # TODO: batch the calls
[docs]
class DbPlan:
"""
Base class containing all possible configuration settings for installed forms. Extensions overwrite these
attributes to set extension-specific settings.
"""
def __init__(self, allow_prompts=True):
self.allow_prompts: bool = allow_prompts
self.disallow_add_to_nav: bool = False
self.send_account_list: bool = True
self.send_account_details: SendAccountDetailsOption = (
SendAccountDetailsOption.SEND_FROM_CACHE
)
self.account_rights_bit_flag: int = 0
self.payload_stored_proc: Optional[str] = None
self.insight_features: list[str] = []
self.marketplace_features: list[str] = []
self.user_property_feature: UserPropertyFeature = (
UserPropertyFeature.FeatureGroupCustUser
)
self.api_stored_procs: list[ApiStoredProc] = []
self.audit_actions: list[AuditAction] = []
self.audit_category: list[AuditCategory] = []
self.data_feeds: list[DataFeed] = []
self.disclaimers: list[Disclaimer] = []
self.gt_data_elements: list[GTDataElement] = []
self.gt_flavor: list[GtFlavor] = []
self.host_account_data_elements: list[HostAccountDataElement] = []
self.host_tran_codes: list[HostTranCode] = []
self.host_tran_code_groups: list[HostTranCodeGroup] = []
self.product_types: list[ProductType] = []
self.products: list[Product] = []
self.report_info: list[ReportInfo] = []
self.required_forms: list[RequiredForm] = []
self.third_party_data_elements: list[ThirdPartyDataElement] = []
self.ui_config_property_data: list[UIConfigPropertyData] = []
self.ui_text_elements: list[UiTextElement] = []
self.user_account_attributes: list[UserAccountAttribute] = []
self.user_property_data: list[UserPropertyData] = []
self.user_property_data_elements: list[UserPropertyDataElement] = []
self.admin_user_property_data_elements: list[AdminUserPropertyDataElement] = []
self.uux_payload: list[UUXPayload] = []
self.vault_info: list[VaultInfo] = []
self.vendor_address: list[VendorAddress] = []
self.vendor_config: list[VendorConfig] = []
self.vendors: list[Vendor] = []
self.ui_text_prefix: Optional[str] = None
self.found_unknown_db_plan_attrs = False
self.upde_allow_user_edit: bool = True
self.upde_allow_user_view: bool = True
self.upde_allow_cust_view: bool = True
def __json__(self):
this_dict = {}
for key, value in vars(self).items():
if value:
if isinstance(value, UserPropertyFeature):
this_dict[key] = value.value
else:
this_dict[key] = value
elif isinstance(value, int):
this_dict[key] = value
return this_dict
@property
def install_steps(self) -> list[InstallStep]:
items_to_return = []
for item in vars(self).values():
if (
isinstance(item, list)
and item != []
and isinstance(item[0], InstallStep)
):
items_to_return.extend(item)
items_to_return = sorted(items_to_return, key=lambda x: x.install_order)
return items_to_return
@property
def install_steps_custom(self) -> list[InstallStep]:
return [item for item in self.install_steps if not item.is_internal]
def prompt_for_editables(
self, steps_to_install: Optional[list[InstallStep]] = None
) -> None:
total_editables = len([x for x in self.install_steps if x.has_editable_attrs])
count = 0
if steps_to_install is None:
steps_to_install = self.install_steps
for step in steps_to_install:
if step.has_editable_attrs:
var_type = step.__class__.__name__
count += 1
_str = ""
for editable_attr in step.editable_attrs:
_str += "{}, ".format(editable_attr)
textui.puts(
f"\n{var_type} row {count} of {total_editables}: (editable columns:"
f" {_str[:-2]})"
)
self._prompt_for_install_step(step)
@staticmethod
def _prompt_for_install_step(install_step: InstallStep) -> InstallStep:
if install_step:
install_step.print_install_vars()
for editable_attr in install_step.editable_attrs:
default = getattr(install_step, editable_attr).value
if len(str(default)) > 150:
textui.puts(
textui.colored.yellow(
f"{editable_attr} value is too long to display/edit in SDK CLI. Default value will be used."
)
)
continue
if isinstance(default, dict):
prompted_default = json.dumps(default)
else:
prompted_default = str(default)
new_val = textui.query(
"{}:".format(editable_attr), default=prompted_default
)
# consider using the InstallStepAttribute.is_required property here
if isinstance(default, dict):
new_val = json.loads(new_val)
if default is None:
if new_val == "None":
new_val = None
edited_attr = getattr(install_step, editable_attr)
edited_attr.value = new_val
return install_step
[docs]
async def install(
self,
logger: Q2LoggerType,
hq_credentials: HqCredentials,
extension_name: str,
custom_only: bool = False,
base_assets_url: Optional[str] = None,
force_update: bool = False,
is_central=False,
) -> None:
"""
Install all items, unless custom_only flag is set to true. In which case only
the custom install_steps will be run
"""
if custom_only:
steps_to_install = self.install_steps_custom
else:
steps_to_install = self.install_steps
if self.allow_prompts:
self.prompt_for_editables(steps_to_install)
if steps_to_install:
textui.puts(
textui.colored.yellow(
f"Running {len(steps_to_install)} step(s) of DbPlan"
)
)
ui_text_install_order = 10
ui_text_objs = []
for item in steps_to_install:
item.hq_credentials = hq_credentials
item.logger = logger
item.extension_name = extension_name
if isinstance(item, UIConfigPropertyData):
await item.install(is_central=is_central)
if isinstance(item, UUXPayload):
item.base_override = base_assets_url
if isinstance(item, UiTextElement):
ui_text_objs.append(
UiTextModel(
item.short_name.value,
item.description.value,
item.text_value.value,
device_id=item.device_id.value,
language=item.language.value,
prefix=self.ui_text_prefix,
)
)
ui_text_install_order = item.install_order
else:
if ui_text_objs and item.install_order > ui_text_install_order:
await UiText(logger, hq_credentials).create_bulk(ui_text_objs)
ui_text_objs = []
try:
await item.install()
except DevOnlyError:
logger.debug(
"db plan install marked as dev only. CDM will install in a different fashion."
)
if ui_text_objs:
if force_update:
await force_ui_text_updates(logger, hq_credentials, ui_text_objs)
else:
await UiText(logger, hq_credentials).create_bulk(ui_text_objs)
await self._update_form_specific_items(logger, hq_credentials, extension_name)
async def _update_form_specific_items(
self, logger: Q2LoggerType, hq_credentials: HqCredentials, extension_name: str
) -> None:
form_obj = Form(logger, hq_credentials)
try:
existing_form = await form_obj.get_by_name(extension_name)
except DatabaseDataError:
return
existing_config = existing_form.findtext("Config") or "{}"
await form_obj.update(
existing_form.ShortName.text,
existing_form.Url.text,
json.loads(existing_config),
existing_form.AccountRightsBitFlag.pyval,
DbPlanRequirements(
self.send_account_list,
self.send_account_details,
self.payload_stored_proc,
existing_form.AccountRightsBitFlag.pyval,
),
)
[docs]
async def uninstall(
self,
logger: Q2LoggerType,
hq_credentials: HqCredentials,
extension_name: str,
custom_only: bool = False,
) -> None:
"""
Remove all items, unless custom_only flag is set to true. In which case only
the custom install_steps will be removed
"""
ui_text_objs = []
if not custom_only:
install_steps = self.install_steps
else:
install_steps = self.install_steps_custom
for item in reversed(install_steps):
item.hq_credentials = hq_credentials
item.logger = logger
item.extension_name = extension_name
if isinstance(item, UiTextElement):
ui_text_objs.append(
UiTextModel(
item.short_name.value,
item.description.value,
item.text_value.value,
device_id=item.device_id.value,
language=item.language.value,
prefix=self.ui_text_prefix,
)
)
else:
await item.uninstall()
if ui_text_objs:
try:
await UiText(logger, hq_credentials).delete_bulk(ui_text_objs)
except DevOnlyError:
logger.info("Skipping UiText delete bulk because we are not in dev")
def to_json(self) -> dict:
return json.loads(RecursiveEncoder().encode(self))
@staticmethod
def get_attr_mapping() -> dict[str, Type[InstallStep]]:
return {
"api_stored_procs": ApiStoredProc,
"audit_actions": AuditAction,
"audit_category": AuditCategory,
"data_feeds": DataFeed,
"disclaimers": Disclaimer,
"gt_data_elements": GTDataElement,
"gt_flavor": GtFlavor,
"host_account_data_elements": HostAccountDataElement,
"host_tran_codes": HostTranCode,
"host_tran_code_groups": HostTranCodeGroup,
"products": Product,
"product_types": ProductType,
"report_info": ReportInfo,
"required_forms": RequiredForm,
"tecton_config": TectonConfig,
"third_party_data_elements": ThirdPartyDataElement,
"ui_config_property_data": UIConfigPropertyData,
"ui_text_elements": UiTextElement,
"user_account_attributes": UserAccountAttribute,
"user_property_data": UserPropertyData,
"user_property_data_elements": UserPropertyDataElement,
"admin_user_property_data_elements": AdminUserPropertyDataElement,
"uux_payload": UUXPayload,
"vault_info": VaultInfo,
"vendor_address": VendorAddress,
"vendor_config": VendorConfig,
"vendors": Vendor,
}
[docs]
def hydrate(self, updated_install_vars_json: list) -> None:
"""
Create install vars class from a json string
:param updated_install_vars_json: List of dictionaries that will be turned into InstallStep objects
"""
attr_mappings = self.get_attr_mapping()
for single_var_as_json in updated_install_vars_json:
var_type = single_var_as_json["var_type"]
obj_vars = vars(self)
for i, var_contents in enumerate(single_var_as_json["contents"]):
install_class = attr_mappings.get(var_type)
attr_in_obj = obj_vars.get(var_type)
if isinstance(attr_in_obj, list):
if len(attr_in_obj) - 1 >= i:
install_class = obj_vars.get(var_type)[i].__class__
if not install_class:
self.found_unknown_db_plan_attrs = True
install_step = get_install_step(self, var_type)
for content_key in var_contents:
setattr(
install_step,
content_key,
create_install_step_attr(var_contents[content_key]),
)
setattr(self, var_type, [install_step])
continue
kwargs = {}
for attribute_name, value_as_json in var_contents.items():
if isinstance(value_as_json, dict):
kwargs[attribute_name] = value_as_json["value"]
else:
kwargs[attribute_name] = value_as_json
db_plan_attr = obj_vars[var_type]
if (len(db_plan_attr) - 1) < i:
db_plan_attr.append(None)
db_plan_attr[i] = install_class(**kwargs)
[docs]
@classmethod
def from_json(cls, db_plan_json: dict) -> "DbPlan":
"""
A compliment to the ``to_json`` method. The JSON that `to_json`` outputs is not
compatible with what the ``hydrate`` method accepts. ``from_json``, however, can
be called with the output of ``to_json``.
"""
new_db_plan = cls()
attr_mappings = cls.get_attr_mapping()
for key, value in db_plan_json.items():
if key in SCALAR_KEYS:
setattr(new_db_plan, key, value)
continue
if key == "user_property_feature":
new_db_plan.user_property_feature = UserPropertyFeature(value)
continue
if key == "uux_payload":
new_db_plan.uux_payload = [
UUXPayload(item["scripts"]) for item in value
]
continue
if key == "vault_info":
vault_infos = []
for item in value:
vault_info_items = {}
for item_key, item_val in item.items():
if item_key != "basename_transforms":
vault_info_items.update({item_key: item_val["value"]})
else:
vault_info_items["basename_transforms"] = item_val
vault_infos.append(VaultInfo(**vault_info_items))
new_db_plan.vault_info = vault_infos
continue
items = []
for val_dict in value:
item = {}
for val_key, val_val in val_dict.items():
item[val_key] = val_val["value"]
items.append(item)
if key in attr_mappings:
value = [attr_mappings[key](**item) for item in items]
else:
value = []
for item in db_plan_json[key]:
install_step = get_install_step(new_db_plan, key)
for content_key in item:
setattr(
install_step,
content_key,
create_install_step_attr(item[content_key]),
)
value.append(install_step)
new_db_plan.found_unknown_db_plan_attrs = True
setattr(new_db_plan, key, value)
return new_db_plan