from argparse import _SubParsersAction
from functools import partial
from typing import List
from dataclasses import dataclass, fields
from lxml.objectify import IntElement, StringElement, fromstring
from lxml import etree, objectify
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.tools.decorators import dev_only
from q2_sdk.hq.table_row import TableRow
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase
[docs]
@dataclass
class HadeUpdateData:
HadeName: str
HadeValue: str
ClickValue: str = None
ClickRequestData: str = None
[docs]
def to_dict(self):
this_as_dict = {}
for param in fields(self):
value = getattr(self, param.name)
if value is not None:
this_as_dict[param.name] = value
return this_as_dict
[docs]
class HostAccountDataRow(RepresentationRowBase):
HostAccountID: IntElement = "HostAccountID"
HADE_ID: IntElement = "HADE_ID"
DataValue: StringElement = "DataValue"
ClickValue: StringElement = "ClickValue"
ClickRequestData: StringElement = "ClickRequestData"
[docs]
class BspHostAccountDataRow(RepresentationRowBase):
HADEName: StringElement = "HADEName"
DataValue: StringElement = "DataValue"
ClickValue: StringElement = "ClickValue"
ClickRequestData: StringElement = "ClickRequestData"
[docs]
class HostAccountData(DbObject):
REPRESENTATION_ROW_CLASS = HostAccountDataRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("add_host_account_data")
subparser.set_defaults(parser="add_host_account_data")
subparser.set_defaults(func=partial(self.create, serialize_for_cli=True))
subparser.add_argument(
"host_account_id", type=int, help="Q2_HostAccount.HostAccountID"
)
subparser.add_argument(
"hade_id", type=int, help="Q2_HostAccountDataElements.HADE_ID"
)
subparser.add_argument("value", help="Value for the HADE")
subparser = parser.add_parser("get_host_account_data")
subparser.set_defaults(parser="get_host_account_data")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser.add_argument(
"host_account_id", type=int, help="Q2_HostAccount.HostAccountID"
)
[docs]
@dev_only
async def create(
self, host_account_id: int, hade_id: int, value: str, serialize_for_cli=False
) -> List[HostAccountDataRow]:
"""Note: this only works in the dev environment"""
assert sum([x for x in [host_account_id, hade_id]]), (
"Please supply a valid host account ID or HADE ID"
)
assert isinstance(value, str), "Please supply a valid value"
response = await self.call_hq(
"sdk_AddHostAccountData",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"host_account_id",
host_account_id,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "hade_id", hade_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar, "value", value
),
]),
)
if serialize_for_cli:
columns = ["HostAccountID", "HADE_ID", "DataValue"]
response = self.serialize_for_cli(response, columns)
return response
[docs]
async def get(
self, host_account_id, overlay_rt_table=True, serialize_for_cli=False
) -> List[BspHostAccountDataRow]:
"""
Get rows from the Q2_HostAccountData table for given host_account_id. By default this overlays data from the Q2_HostAccountDataRT table,
which is used for certain batch processing scenarios. Shouldn't be harmful to leave on all the time.
:param host_account_id: Matches the Q2_HostAccountData.HostAccountID column
:param overlay_rt_table: If True, will also query and overlay the Q2_HostAccountDataRT table
:param serialize_for_cli: Used when running from the command line
"""
async def call_proc(proc_name) -> List[BspHostAccountDataRow]:
return await self.call_hq(
proc_name,
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"HostAccountID",
host_account_id,
)
]),
specific_table="DATA",
)
overlay_hq_response = []
if overlay_rt_table is True:
overlay_hq_response = await call_proc("bsp_GetHostAccountData_RT")
if overlay_hq_response:
overlay_hq_response = fromstring(str(overlay_hq_response[0]))
if self.ret_table_obj:
overlay_hq_response = fromstring(overlay_hq_response.text)
hq_response = await call_proc("bsp_GetHostAccountData")
if not hq_response and not overlay_hq_response:
return []
xml_response = fromstring(str(hq_response[0]))
if self.ret_table_obj:
xml_response = fromstring(xml_response.text)
columns = ["HADEName", "DataValue", "ClickRequestData", "ClickValue"]
response = self._attrs_to_nodes(xml_response.getchildren(), columns)
if len(overlay_hq_response):
for overlay_row in self._attrs_to_nodes(
overlay_hq_response.getchildren(), columns
):
for i, row in enumerate(response):
if row.findtext("HADEName") == overlay_row.findtext("HADEName"):
response[i] = overlay_row
break
else:
response.append(overlay_row)
if serialize_for_cli:
response = self.serialize_for_cli(response, columns)
if not self.ret_table_obj:
return response
else:
rows = [
TableRow(
objectify.fromstring(etree.tostring(lxml_element)),
BspHostAccountDataRow,
)
for lxml_element in response
]
return rows
@staticmethod
def _attrs_to_nodes(nodes, attr_names):
collected_nodes = []
for node in nodes:
row = etree.Element("row")
for attr in attr_names:
value = node.get(attr)
if value:
etree.SubElement(row, attr).text = value
collected_nodes.append(row)
return collected_nodes
[docs]
async def update(self, host_account_id, hade_values: list[HadeUpdateData]):
root = etree.Element("ImportRows")
for hade in hade_values:
etree.SubElement(
root, "ImportRow", HostAccountId=str(host_account_id), **hade.to_dict()
)
xml_string = etree.tostring(root, encoding="unicode")
hq_response = await self.call_hq(
"bsp_UpdateHade_BT",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Xml, "data", xml_string
)
]),
)
return hq_response
[docs]
async def delete(self, hade_id: int, host_account_id: int):
response = await self.call_hq(
"sdk_RemoveHostAccountData",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "hadeId", hade_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"hostAccountId",
host_account_id,
),
]),
)
return response