from typing import Union, Unpack
from q2_sdk.core.install_steps.base import (
InstallStep,
InstallStepAttribute,
InstallStepArguments,
)
from q2_sdk.core.exceptions import DatabaseDataError
from q2_sdk.hq.db.host_tran_code import DOrC, HostTranCode as HostTranCodeDbObj
[docs]
class HostTranCode(InstallStep):
"""Interacts with the Q2_HostTranCode table"""
def __init__(
self,
host_tran_code: str,
tran_code_group_name: str,
d_or_c: Union[DOrC, str],
tran_code_desc: str,
ofx_transaction_type_name: str,
show_image=False,
transaction_type=0,
ofx_check_num_handling=0,
show_transaction_item_detail=0,
**kwargs: Unpack[InstallStepArguments],
):
super().__init__(**kwargs)
self.install_order = 20
if isinstance(d_or_c, DOrC):
d_or_c = d_or_c.value
self.host_tran_code = InstallStepAttribute(host_tran_code)
self.tran_code_group_name = InstallStepAttribute(tran_code_group_name)
self.d_or_c = InstallStepAttribute(d_or_c)
self.tran_code_desc = InstallStepAttribute(tran_code_desc)
self.ofx_transaction_type_name = InstallStepAttribute(ofx_transaction_type_name)
self.show_image = InstallStepAttribute(show_image, is_bool=True)
self.transaction_type = InstallStepAttribute(transaction_type)
self.ofx_check_num_handling = InstallStepAttribute(ofx_check_num_handling)
self.show_transaction_item_detail = InstallStepAttribute(
show_transaction_item_detail
)
[docs]
async def install(self):
await super().install()
db_obj = HostTranCodeDbObj(self.logger, hq_credentials=self.hq_credentials)
already_exists = False
try:
await db_obj.get_by_name(self.host_tran_code.value)
already_exists = True
except DatabaseDataError:
pass
if already_exists:
return
await db_obj.create(
self.host_tran_code.value,
self.tran_code_group_name.value,
self.d_or_c.value,
self.tran_code_desc.value,
self.ofx_transaction_type_name.value,
show_image=self.show_image.value,
transaction_type=self.transaction_type.value,
ofx_check_num_handling=self.ofx_check_num_handling.value,
show_transaction_item_detail=self.show_transaction_item_detail.value,
)
[docs]
async def uninstall(self):
db_obj = HostTranCodeDbObj(self.logger, self.hq_credentials)
await db_obj.delete(self.host_tran_code.value)