from typing import Optional, Unpack
from q2_sdk.core.install_steps.base import (
InstallStep,
InstallStepAttribute,
InstallStepArguments,
)
from q2_sdk.hq.db.product import Product as ProductDbObj
[docs]
class Product(InstallStep):
"""Interacts with the Q2_products table"""
def __init__(
self,
product_name: str,
host_product_code: str,
tran_code_group_name: str,
product_type_name: str,
hydra_product_code: str,
product_nick_name: str,
voice_file: Optional[str] = None,
vendor_name: Optional[str] = None,
product_type_id: Optional[int] = None,
allow_open=False,
allow_close=False,
**kwargs: Unpack[InstallStepArguments],
):
super().__init__(**kwargs)
self.product_name = InstallStepAttribute(product_name)
self.host_product_code = InstallStepAttribute(host_product_code)
self.tran_code_group_name = InstallStepAttribute(tran_code_group_name)
self.product_type_name = InstallStepAttribute(product_type_name)
self.hydra_product_code = InstallStepAttribute(hydra_product_code)
self.product_nick_name = InstallStepAttribute(product_nick_name)
self.voice_file = InstallStepAttribute(voice_file)
self.vendor_name = InstallStepAttribute(vendor_name)
self.product_type_id = InstallStepAttribute(product_type_id, is_editable=True)
self.allow_open = InstallStepAttribute(allow_open, is_bool=True)
self.allow_close = InstallStepAttribute(allow_close, is_bool=True)
self.install_order = 20
[docs]
async def install(self):
await super().install()
db_obj = ProductDbObj(self.logger, hq_credentials=self.hq_credentials)
existing = await db_obj.get_list_by_name(self.product_name.value)
if existing:
return
await db_obj.create(
self.product_name.value,
self.host_product_code.value,
self.tran_code_group_name.value,
self.product_type_name.value,
self.hydra_product_code.value,
self.product_nick_name.value,
voice_file=self.voice_file.value,
vendor_name=self.vendor_name.value,
product_type_id=self.product_type_id.value,
allow_open=self.allow_open.value,
allow_close=self.allow_close.value,
)
[docs]
async def uninstall(self):
db_obj = ProductDbObj(self.logger, self.hq_credentials)
product_list = await db_obj.get_list_by_name(self.product_name.value)
for product in product_list:
await db_obj.delete(product.ProductID.pyval)