from typing import Optional, Unpack
from q2_sdk.core.install_steps.base import (
InstallStep,
InstallStepAttribute,
InstallStepArguments,
)
from q2_sdk.core.exceptions import InstallStepError
from q2_sdk.hq.db.user_property_data import UserPropertyData as UserPropertyDataDbObj
from q2_sdk.hq.db.product import Product as ProductDbObj
from q2_sdk.hq.db.product_type import ProductType as ProductTypeDbObj
[docs]
class UserPropertyData(InstallStep):
"""Interacts with the Q2_UserPropertyData table"""
def __init__(
self,
property_name,
property_value,
user_id=None,
group_id=None,
customer_id=None,
host_account_id=None,
ui_source=None,
product_name=None,
product_type_name=None,
host_product_code: Optional[str] = None,
**kwargs: Unpack[InstallStepArguments],
):
super().__init__(**kwargs)
self.install_order = 20
self.property_name = InstallStepAttribute(property_name)
self.property_value = InstallStepAttribute(property_value, is_editable=True)
self.user_id = InstallStepAttribute(user_id)
self.group_id = InstallStepAttribute(group_id)
self.customer_id = InstallStepAttribute(customer_id)
self.host_account_id = InstallStepAttribute(host_account_id)
self.ui_source = InstallStepAttribute(ui_source)
self.product_name = InstallStepAttribute(product_name)
self.product_type_name = InstallStepAttribute(product_type_name)
self.host_product_code = InstallStepAttribute(host_product_code)
[docs]
async def install(self):
await super().install()
product_type_id = await self._get_product_type_id_from_name(
self.product_type_name.value
)
product_id_list = await self._get_product_id_list_from_name()
if product_id_list:
for product in product_id_list:
await self.call_proc(
product_id=product.ProductID.text, product_type_id=product_type_id
)
else:
await self.call_proc(product_type_id=product_type_id)
[docs]
async def uninstall(self):
await UserPropertyDataDbObj(
self.logger, hq_credentials=self.hq_credentials
).delete_all_property_data(self.property_name.value)
async def call_proc(self, product_id=None, product_type_id=None):
db_obj = UserPropertyDataDbObj(self.logger, hq_credentials=self.hq_credentials)
existing = await db_obj.get(
self.property_name.value,
user_id=self.user_id.value,
group_id=self.group_id.value,
customer_id=self.customer_id.value,
host_account_id=self.host_account_id.value,
ui_source=self.ui_source.value,
product_id=product_id,
product_type_id=product_type_id,
)
if not len(existing):
await db_obj.create(
self.property_name.value,
self.property_value.value,
user_id=self.user_id.value,
group_id=self.group_id.value,
customer_id=self.customer_id.value,
host_account_id=self.host_account_id.value,
ui_source=self.ui_source.value,
product_id=product_id,
product_type_id=product_type_id,
)
async def _get_product_id_list_from_name(self):
if not self.product_name.value:
return None
product_obj = ProductDbObj(self.logger, self.hq_credentials)
product_id_list = await product_obj.get_list_by_name(
self.product_name.value,
self.product_type_name.value,
self.host_product_code.value,
)
if not product_id_list:
error_msg = (
f'No Product called "{self.product_name.value}" with '
f'Host Product Code "{self.host_product_code.value}" and '
f'Product Type ID "{self.product_type_name.value}" in database'
)
self.logger.error(error_msg)
raise InstallStepError(error_msg)
return product_id_list
async def _get_product_type_id_from_name(self, product_type_name):
if not product_type_name:
return None
product_type_obj = ProductTypeDbObj(self.logger, self.hq_credentials)
product_type = await product_type_obj.get_by_name(self.product_type_name.value)
if product_type is None:
error_msg = (
f'No ProductType called "{self.product_type_name.value}" in database'
)
self.logger.error(error_msg)
raise InstallStepError(error_msg)
return product_type.ProductTypeID.text