from argparse import _SubParsersAction
from functools import partial
from typing import List
from lxml.objectify import IntElement, StringElement
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.hq.models.online_session import OnlineSession
from q2_sdk.hq.models.online_user import OnlineUser
from q2_sdk.hq.models.hq_params.stored_procedure import Param
from .db_object import DbObject
from .audit_record import AuditRecord
from .representation_row_base import RepresentationRowBase
D_TYPES = ExecuteStoredProcedure.DataType
[docs]
class CustomerDataRow(RepresentationRowBase):
DataID: IntElement = "DataID"
ShortName: StringElement = "ShortName"
CustomerID: IntElement = "CustomerID"
GTDataValue: StringElement = "GTDataValue"
FileID: IntElement = "FileID"
CreateDate: StringElement = "CreateDate"
DisplayName: StringElement = "DisplayName"
[docs]
class CustomerData(DbObject):
"""
Allows for DB storage on a per customer basis. Combines the Q2_CustomerData and Q2_CustomerDataElements tables
"""
REPRESENTATION_ROW_CLASS = CustomerDataRow
RESPONSE_FIELDS = [
"DataID",
"ShortName",
"CustomerID",
"GTDataValue",
"FileID",
"CreateDate",
"DisplayName",
]
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_customer_data")
subparser.set_defaults(parser="get_customer_data")
subparser.add_argument("customer_id", type=int, help="Q2_User.CustomerID")
subparser.add_argument("short_name", help="Q2_CustomerDataElement.ShortName")
subparser.add_argument(
"--no-trunc", action="store_true", help="Do not truncate GTDataValue"
)
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser = parser.add_parser("get_customer_data_all")
subparser.set_defaults(parser="get_customer_data_all")
subparser.add_argument("customer_id", type=int, help="Q2_User.CustomerID")
subparser.add_argument(
"--no-trunc", action="store_true", help="Do not truncate GTDataValue"
)
subparser.set_defaults(func=partial(self.get_multi, serialize_for_cli=True))
subparser = parser.add_parser("get_customer_data_by_value")
subparser.set_defaults(parser="get_customer_data_by_value")
subparser.add_argument("short_name", help="Q2_CustomerDataElement.ShortName")
subparser.add_argument("value", help="Q2_CustomerDataElement.GTDataValue")
subparser.add_argument(
"--no-trunc", action="store_true", help="Do not truncate GTDataValue"
)
subparser.set_defaults(func=partial(self.get_by_value, serialize_for_cli=True))
subparser = parser.add_parser("get_customer_data_by_short_name")
subparser.set_defaults(parser="get_customer_data_by_short_name")
subparser.add_argument("short_name", help="Q2_CustomerDataElement.ShortName")
subparser.add_argument(
"--no-trunc", action="store_true", help="Do not truncate GTDataValue"
)
subparser.set_defaults(
func=partial(self.get_by_short_name, serialize_for_cli=True)
)
subparser = parser.add_parser("add_customer_data")
subparser.set_defaults(parser="add_customer_data")
subparser.set_defaults(func=partial(self.create))
subparser.add_argument(
"customer_id", type=int, help="Q2_CustomerData.CustomerID"
)
subparser.add_argument("short_name", help="Q2_CustomerDataElement.ShortName")
subparser.add_argument("value", help="Q2_CustomerData.GTDataValue")
subparser = parser.add_parser("remove_customer_data")
subparser.set_defaults(parser="remove_customer_data")
subparser.set_defaults(
func=partial(
self.delete, online_session=OnlineSession(), online_user=OnlineUser()
)
)
subparser.add_argument(
"customer_id", type=int, help="Q2_CustomerData.CustomerID"
)
subparser.add_argument("data_id", help="Q2_CustomerData.DataID")
subparser = parser.add_parser("update_customer_data")
subparser.set_defaults(parser="update_customer_data")
subparser.set_defaults(func=partial(self.update))
subparser.add_argument(
"customer_id", type=int, help="Q2_CustomerData.CustomerID"
)
subparser.add_argument("short_name", help="Q2_CustomerDataElement.ShortName")
subparser.add_argument("value", help="Q2_CustomerData.GTDataValue")
[docs]
async def get(
self, customer_id: int, short_name: str, no_trunc=False, serialize_for_cli=False
) -> List[CustomerDataRow]:
"""
Gets the specific customer data information for the given customer id and short name
:param customer_id: customer identifier
:param short_name: customer data short name
:param no_trunc: bool that will truncate the data value if it is too long to display in the CLI
:param serialize_for_cli: Used when running from the command line
:return: customer data value for the given inputs
"""
truncate = not no_trunc
sql_params = []
Param(str(customer_id), D_TYPES.Int, "customerID").add_to_param_list(sql_params)
Param(short_name, D_TYPES.VarChar, "dataElement").add_to_param_list(sql_params)
response = await self.call_hq(
"Q2_GetCustomerData", ExecuteStoredProcedure.SqlParameters(sql_params)
)
fields_to_truncate = []
if serialize_for_cli:
if truncate:
fields_to_truncate = ["GTDataValue"]
response = self.serialize_for_cli(
response, self.RESPONSE_FIELDS, fields_to_truncate=fields_to_truncate
)
return response
[docs]
async def get_by_short_name(
self, short_name: str, no_trunc=False, serialize_for_cli=False
) -> List[CustomerDataRow]:
"""This will get all rows that have this short_name available in Q2_CustomerData"""
truncate = not no_trunc
sql_params = []
Param(short_name, D_TYPES.VarChar, "short_name").add_to_param_list(sql_params)
response = await self.call_hq(
"sdk_GetCustomerDataByShortName",
ExecuteStoredProcedure.SqlParameters(sql_params),
)
fields_to_truncate = []
if serialize_for_cli:
if truncate:
fields_to_truncate = ["GTDataValue"]
response = self.serialize_for_cli(
response, self.RESPONSE_FIELDS, fields_to_truncate=fields_to_truncate
)
return response
[docs]
async def get_by_value(
self, short_name: str, value: str, no_trunc=False, serialize_for_cli=False
) -> List[CustomerDataRow]:
truncate = not no_trunc
sql_params = []
Param(value, D_TYPES.VarChar, "value").add_to_param_list(sql_params)
Param(short_name, D_TYPES.VarChar, "dataElement").add_to_param_list(sql_params)
response = await self.call_hq(
"sdk_GetCustomerDataByValue",
ExecuteStoredProcedure.SqlParameters(sql_params),
)
if serialize_for_cli:
fields_to_truncate = []
if truncate:
fields_to_truncate = ["GTDataValue"]
response = self.serialize_for_cli(
response, self.RESPONSE_FIELDS, fields_to_truncate=fields_to_truncate
)
return response
[docs]
async def get_multi(
self, customer_id: int, serialize_for_cli=False, no_trunc=False
) -> List[CustomerDataRow]:
"""
Returns all customer data associated with the given customer id
:param customer_id: customer identifier
:return: customer data values for the associated customer
"""
truncate = not no_trunc
sql_params = []
Param(str(customer_id), D_TYPES.Int, "customer_id").add_to_param_list(
sql_params
)
response = await self.call_hq(
"sdk_GetCustomerDataByID", ExecuteStoredProcedure.SqlParameters(sql_params)
)
if serialize_for_cli:
fields_to_truncate = []
if truncate:
fields_to_truncate = ["GTDataValue"]
response = self.serialize_for_cli(
response, self.RESPONSE_FIELDS, fields_to_truncate=fields_to_truncate
)
return response
[docs]
async def create(self, customer_id: int, short_name: str, value: str):
sql_params = []
Param(str(customer_id), D_TYPES.Int, "customerID").add_to_param_list(sql_params)
Param(short_name, D_TYPES.VarChar, "dataElement").add_to_param_list(sql_params)
Param(value, D_TYPES.VarChar, "data").add_to_param_list(sql_params)
ten_kb = 1024 * 10
if len(str(value)) > ten_kb:
self.logger.warning(
"Values larger than 10Kb in this table can lead to excessive "
"load on the database at high volumes. Please consider "
"entering a key into this table and using it to query another "
"system"
)
return await self.call_hq(
"Q2_AddUpdateCustomerData", ExecuteStoredProcedure.SqlParameters(sql_params)
)
[docs]
async def update(self, customer_id: int, short_name: str, value: str):
return await self.create(customer_id, short_name, value)
[docs]
async def delete(
self,
customer_id: int,
data_id: int,
online_session: OnlineSession,
online_user: OnlineUser,
):
sql_params = []
Param(str(customer_id), D_TYPES.Int, "customer_id").add_to_param_list(
sql_params
)
Param(str(data_id), D_TYPES.Int, "data_id").add_to_param_list(sql_params)
result = await self.call_hq(
"sdk_RemoveCustomerData", ExecuteStoredProcedure.SqlParameters(sql_params)
)
if self.hq_response.success:
audit = AuditRecord(self.logger, self.hq_credentials)
await audit.create(
f"CustomerData deleted. customer_id: {customer_id} data_id: {data_id}",
online_session.session_id,
workstation_id=online_session.workstation,
customer_id=online_user.customer_id,
user_id=online_user.user_id,
user_logon_id=online_user.user_logon_id,
)
return result