from dataclasses import dataclass
from typing import Optional, Unpack, Union
from q2_sdk.core.configuration import settings
from q2_sdk.core.install_steps.base import (
InstallStep,
InstallStepArguments,
InstallStepAttribute,
)
from q2_sdk.core.install_steps.host_tran_code_group import HostTranCodeGroup
from q2_sdk.core.install_steps.product_type import ProductType
from q2_sdk.core.install_steps.product import Product
from q2_sdk.entrypoints.invalidate_hq_cache import invalidate
from q2_sdk.hq.db.db_object import DatabaseDataError
from q2_sdk.hq.db.host_account import HostAccount
from q2_sdk.hq.db.host_account_data_element import (
CreateHADEParams,
HADEDataTypes,
HostAccountDataElement,
)
from q2_sdk.hq.db.linked_account import LinkedAccount
from q2_sdk.hq.db.product_type import AprOrApy, HydraProductTypeCode
from q2_sdk.hq.db.product_type import ProductType as ProductTypeDbObj
from q2_sdk.hq.db.system_property_data_with_references import (
SystemPropertyDataWithReferences,
)
from q2_sdk.hq.db.user import User
[docs]
@dataclass
class ProductTypeConfig:
"""Configuration for creating a ProductType."""
product_type_name: str
apr_or_apy: Union[AprOrApy, str]
hydra_product_type_code: Union[HydraProductTypeCode, str]
host_product_type_code: str # shortname that HQ uses for identity
is_external: bool = False # If True, this is an account not found on the core
[docs]
@dataclass
class ProductConfig:
"""Configuration for creating a Product."""
product_name: str
host_product_code: str
tran_code_group_name: str
hydra_product_code: str
product_nick_name: Optional[str] = None
voice_file: Optional[str] = None
vendor_name: Optional[str] = None
product_type_id: Optional[int] = None
allow_open: bool = False
allow_close: bool = False
[docs]
@dataclass
class SystemPropertyDataConfig:
hade_to_display_1: str = "AvailBal"
hade_to_display_2: str = "CurBal"
hade_to_display_1_data_type: HADEDataTypes = HADEDataTypes.String
hade_to_display_2_data_type: HADEDataTypes = HADEDataTypes.String
[docs]
@dataclass
class NewAccountConfig:
"""Configuration for creating/linking a new account."""
account_internal: str
account_external: str
cif: str
link_login: str
account_description: str = "Test Account"
is_external: bool = False
link_access: int = 7
[docs]
class AccountDetails(InstallStep):
"""
Combined pre-requisites for account details adapter.
This install step creates the HostTranCodeGroup, ProductType, Product,
and SystemPropertyData entries required for an account details adapter
in a single db_plan entry.
"""
def __init__(
self,
product_type_config: ProductTypeConfig,
product_config: ProductConfig,
# SystemPropertyData fields for HadeToDisplay
system_property_data_config: SystemPropertyDataConfig,
new_account_config: Optional[NewAccountConfig] = None,
**kwargs: Unpack[InstallStepArguments],
):
super().__init__(**kwargs)
self.product_type_config = InstallStepAttribute(product_type_config)
self.product_config = InstallStepAttribute(product_config)
self.system_property_data_config = InstallStepAttribute(
system_property_data_config
)
self.new_account_config = InstallStepAttribute(new_account_config)
self.host_tran_code_group = InstallStepAttribute(
HostTranCodeGroup(description=product_config.tran_code_group_name)
)
self.hade_to_display_1 = InstallStepAttribute(
system_property_data_config.hade_to_display_1
)
self.hade_to_display_2 = InstallStepAttribute(
system_property_data_config.hade_to_display_2
)
self.hade_to_display_1_data_type = InstallStepAttribute(
system_property_data_config.hade_to_display_1_data_type
)
self.hade_to_display_2_data_type = InstallStepAttribute(
system_property_data_config.hade_to_display_2_data_type
)
self.product_type = InstallStepAttribute(
ProductType(
product_type_name=product_type_config.product_type_name,
apr_or_apy=product_type_config.apr_or_apy,
hydra_product_type_code=(product_type_config.hydra_product_type_code),
host_product_type_code=(product_type_config.host_product_type_code),
is_external=product_type_config.is_external,
)
)
self.product = InstallStepAttribute(
Product(
product_name=product_config.product_name,
host_product_code=product_config.host_product_code,
tran_code_group_name=product_config.tran_code_group_name,
product_type_name=product_type_config.product_type_name,
hydra_product_code=product_config.hydra_product_code,
product_nick_name=product_config.product_nick_name,
voice_file=product_config.voice_file,
vendor_name=product_config.vendor_name,
product_type_id=product_config.product_type_id,
allow_open=product_config.allow_open,
allow_close=product_config.allow_close,
)
)
self.install_order = 20
async def _get_product_type_id(self) -> Optional[int]:
"""Look up the product type ID by name."""
product_type_db = ProductTypeDbObj(
self.logger, hq_credentials=self.hq_credentials, ret_table_obj=True
)
product_types = await product_type_db.get()
for pt in product_types:
if pt.ProductTypeName == self.product_type_config.value.product_type_name:
return pt.ProductTypeID
return None
async def _install_system_property_data(self):
"""Add HadeToDisplay1 and HadeToDisplay2 system property data for the product type."""
product_type_id = await self._get_product_type_id()
if product_type_id is None:
self.logger.warning(
f"Could not find ProductTypeID for "
f"'{self.product_type_config.value.product_type_name}', "
"skipping system property data installation"
)
return
spd = SystemPropertyDataWithReferences(
self.logger, hq_credentials=self.hq_credentials
)
# Check if properties already exist
hade_ids_to_display = []
hade1_obj = HostAccountDataElement(
self.logger, self.hq_credentials, ret_table_obj=True
)
try:
hade1 = await hade1_obj.get_by_name(self.hade_to_display_1.value)
except DatabaseDataError:
self.logger.info(
f"HADE '{self.hade_to_display_1.value}' not found, creating it"
)
await hade1_obj.create(
CreateHADEParams(
hade_name=self.hade_to_display_1.value,
hade_desc=self.hade_to_display_1.value,
hade_data_type=self.hade_to_display_1_data_type.value,
)
)
hade1 = await hade1_obj.get_by_name(self.hade_to_display_1.value)
hade_ids_to_display.append(hade1.HADE_ID)
hade2_obj = HostAccountDataElement(
self.logger, self.hq_credentials, ret_table_obj=True
)
try:
hade2 = await hade2_obj.get_by_name(self.hade_to_display_2.value)
except DatabaseDataError:
self.logger.info(
f"HADE '{self.hade_to_display_2.value}' not found, creating it"
)
await hade2_obj.create(
CreateHADEParams(
hade_name=self.hade_to_display_2.value,
hade_desc=self.hade_to_display_2.value,
hade_data_type=self.hade_to_display_2_data_type.value,
)
)
hade2 = await hade2_obj.get_by_name(self.hade_to_display_2.value)
hade_ids_to_display.append(hade2.HADE_ID)
# property_value = HADE_ID
spde_hades = ["HadeToDisplay1", "HadeToDisplay2"]
for hade_id, spde_hade in zip(hade_ids_to_display, spde_hades):
await spd.add(
property_name=spde_hade,
property_value=hade_id,
product_type_id=product_type_id,
)
# Invalidate SystemProperties cache
await invalidate(
logger=self.logger,
hq_credentials=self.hq_credentials,
refresh_type="SystemProperties",
)
async def _install_new_account(self):
"""Create a new host account and link to a user.
In dev: Creates account if it doesn't exist, then links to user.
In production: Only links existing account to user (cannot create accounts).
"""
if not self.new_account_config.value:
return
config = self.new_account_config.value
host_account = HostAccount(self.logger, self.hq_credentials)
# Check if account already exists
existing = await host_account.search_by_internal_number(config.account_internal)
host_account_id = None
if existing:
host_account_id = int(existing[0].HostAccountID)
self.logger.info(
f"Account '{config.account_internal}' already exists "
f"(HostAccountID: {host_account_id})"
)
elif settings.DEPLOY_ENV == "DEV":
# Create account only in dev
self.logger.info(f"Creating account '{config.account_internal}'")
results = await host_account.create(
account_internal=config.account_internal,
account_external=config.account_external,
host_product_code=self.product_config.value.host_product_code,
host_product_type_code=self.product_type_config.value.host_product_type_code,
cif=config.cif,
account_description=config.account_description,
is_external=config.is_external,
)
host_account_id = int(results[0].HostAccountID.text)
self.logger.info(f"Created account with HostAccountID: {host_account_id}")
else:
self.logger.warning(
f"Account '{config.account_internal}' does not exist and cannot be "
"created in production environment, skipping"
)
return
# Link account to user if specified
if config.link_login and host_account_id:
user_obj = User(self.logger, hq_credentials=self.hq_credentials)
response = await user_obj.get(login_name=config.link_login)
if not response:
self.logger.warning(
f"Could not find user with login '{config.link_login}', "
"skipping account linking"
)
return
user_id = response[0].UserID.text
customer_id = response[0].CustomerID.text
self.logger.info(
f"Linking account to user '{config.link_login}' "
f"with access level {config.link_access}"
)
linked_account = LinkedAccount(self.logger, self.hq_credentials)
result = await linked_account.add(
customer_id=customer_id,
user_id=user_id,
account_access=config.link_access,
host_account_id=host_account_id,
)
if result and "ERROR" in str(result):
self.logger.warning(f"Failed to link account: {result}")
else:
self.logger.info("Successfully linked account to user")
[docs]
async def install(self):
await super().install()
for step in [self.product_type.value, self.product.value]:
step.hq_credentials = self.hq_credentials
step.logger = self.logger
step.extension_name = self.extension_name
await step.install()
await self._install_system_property_data()
await self._install_new_account()
[docs]
async def uninstall(self):
for step in [self.product.value, self.product_type.value]:
step.hq_credentials = self.hq_credentials
step.logger = self.logger
step.extension_name = self.extension_name
await step.uninstall()