import json
from argparse import _SubParsersAction
from functools import partial
from typing import List, Optional
from lxml.objectify import BoolElement, IntElement, StringElement
from q2_sdk.core.exceptions import DatabaseDataError
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from .base import InstallerObj
from .db_object import DEFAULT
from . import form
from .representation_row_base import BaseFormRow
from .system_property_data import SystemPropertyData
from .vendor_address import VendorAddress
from .vendor_config import VendorConfig
[docs]
class VendorRow(BaseFormRow):
VendorID: IntElement = "VendorID"
VendorName: StringElement = "VendorName"
ActionID: IntElement = "ActionID"
PropertyID: IntElement = "PropertyID"
PropertyName: StringElement = "PropertyName"
ParametersBitflag: IntElement = "ParametersBitflag"
VendorAddressID: IntElement = "VendorAddressID"
PurgeAcctListOnSuccess: BoolElement = "PurgeAcctListOnSuccess"
PayloadStoredProc: StringElement = "PayloadStoredProc"
Address: StringElement = "Address"
Description: StringElement = "Description"
TextValue: StringElement = "TextValue"
NavigationNodeID: IntElement = "NavigationNodeID"
HasUserProperties: BoolElement = "HasUserProperties"
BalanceHadeSystemProperty: StringElement = "BalanceHadeSystemProperty"
[docs]
class Vendor(InstallerObj):
GET_BY_NAME_KEY = "VendorName"
NAME = "Vendor"
REPRESENTATION_ROW_CLASS = VendorRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_vendors")
subparser.set_defaults(parser="get_vendors")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser.add_argument(
"--show-only-sso",
default=True,
action="store_false",
dest="show_all",
help="By default, show all vendors",
)
subparser.add_argument(
"--name", dest="short_name", help="Q2_Vendors.VendorName"
)
subparser.add_argument(
"-g",
"--with-groups",
action="store_true",
help="Limit results to a single form",
)
subparser.set_defaults(with_groups=False)
subparser = parser.add_parser("add_vendor")
subparser.set_defaults(parser="add_vendor")
subparser.set_defaults(func=partial(self.create))
subparser.add_argument("vendor_name", help="Vendor Name")
subparser = parser.add_parser("remove_vendor")
subparser.set_defaults(parser="remove_vendor")
subparser.set_defaults(func=partial(self.delete))
subparser.add_argument("vendor_name", help="Vendor Name")
subparser = parser.add_parser("update_vendor")
subparser.set_defaults(parser="update_vendor")
subparser.set_defaults(func=partial(self.update, serialize_for_cli=True))
subparser.add_argument("vendor_id", help="Vendor ID")
subparser.add_argument(
"-vn", "--vendor-name", help="Vendor Name", default=DEFAULT
)
subparser.add_argument(
"-pb",
"--parameters-bit-flag",
help="Q2_Vendors.ParametersBitFlag",
default=DEFAULT,
)
subparser.add_argument(
"-a", "--audit-action-id", help="Q2_AuditAction.ShortName", default=DEFAULT
)
subparser.add_argument(
"-up",
"--property-id",
help="Q2_UserPropertyDataElement.PropertyName",
default=DEFAULT,
)
subparser.add_argument(
"-ps",
"--payload-stored-proc",
help="Q2_Vendors.ParametersBitFlag",
default=DEFAULT,
)
subparser.add_argument(
"-pa",
"--purge-account-list",
action="store_true",
help="Q2_Vendors.PurgeAcctListOnSuccess",
default=False,
)
[docs]
async def get(
self,
serialize_for_cli=False,
short_name: Optional[str] = None,
vendor_id: Optional[int] = None,
show_all=False,
with_groups=False,
) -> List[VendorRow]:
"""
:param serialize_for_cli: Used when running from the command line
:param short_name: Q2_Vendor.VendorName
:param vendor_id: Q2_Vendor.VendorID
:param show_all: By default we only show SSO Vendor entries.
:param with_groups: Include group associations (can be costly for large payloads)
:return:
"""
response = await self.call_hq("sdk_GetVendors")
if not show_all:
response = [x for x in response if x.find("ActionID") and x.find("Address")]
for row in response:
row.IsCustomerCreated = self._is_customer_created(row)
detailed = False
if short_name:
detailed = True
response = [x for x in response if x.VendorName == short_name]
elif vendor_id:
detailed = True
response = [x for x in response if x.VendorID.text == str(vendor_id)]
if with_groups:
response = await form.add_groups_to_forms(
self.logger, self.hq_credentials, response
)
if serialize_for_cli:
response = await self._serialize_get(response, detailed)
return response
[docs]
async def get_by_name(self, name: str, **kwargs) -> VendorRow:
return await super().get_by_name(name, show_all=True, **kwargs)
async def _serialize_get(self, response: List[VendorRow], detailed=False):
if not response:
return ""
if detailed:
vendor = response[0]
vendor_configs = await VendorConfig(self.logger, self.hq_credentials).get(
vendor.VendorName.text
)
config = {x.ConfigName.text: x.ConfigValue.text for x in vendor_configs}
config = json.dumps(config, sort_keys=True, indent=4)
vendor_groups = vendor.findtext("Groups")
response_list = [
"VendorID:\t{}".format(vendor.findtext("VendorID")),
"VendorName:\t{}".format(vendor.findtext("VendorName")),
"Groups:\t{}".format(vendor_groups),
"ActionID:\t{}".format(vendor.findtext("ActionID")),
"ParametersBitflag:\t{}".format(vendor.findtext("ParametersBitflag")),
"IsCustomerCreated\t{}".format(vendor.findtext("IsCustomerCreated")),
"Config:\t{}".format(config),
]
if vendor_groups:
response = "\n".join(response_list)
else:
response_list.remove("Groups:\t{}".format(vendor_groups))
response = "\n".join(response_list)
else:
columns = [
"VendorID",
"VendorName",
"ActionID",
"ParametersBitflag",
"VendorAddressID",
"Address",
"NavigationNodeID",
"HasUserProperties",
"NavNodeName",
"IsCustomerCreated",
"BalanceHadeSystemProperty",
]
response = self.serialize_for_cli(response, columns)
return response
[docs]
async def create(
self,
vendor_name,
vendor_address_id: str = None,
audit_action_id: str = None,
parameter_bit_flag: int = None,
purge_account_list: bool = False,
payload_stored_proc: str = None,
property_name: str = None,
balance_hade_system_property: str = None,
) -> VendorRow:
parameters = [
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar, "vendor_name", vendor_name
)
]
if vendor_address_id:
parameters.append(
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"vendor_address_id",
vendor_address_id,
)
)
if audit_action_id:
parameters.append(
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"audit_action_id",
audit_action_id,
)
)
if parameter_bit_flag:
parameters.append(
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"parameter_bit_flag",
parameter_bit_flag,
)
)
if purge_account_list:
parameters.append(
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Bit,
"purge_account_list",
purge_account_list,
)
)
if payload_stored_proc:
parameters.append(
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"payload_stored_proc",
payload_stored_proc,
)
)
if property_name:
parameters.append(
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"property_name",
property_name,
)
)
if property_name:
parameters.append(
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"balance_hade_system_property",
balance_hade_system_property,
)
)
results = await self.call_hq(
"sdk_AddVendor", ExecuteStoredProcedure.SqlParameters(parameters)
)
return results[0]
[docs]
async def delete(self, vendor_name):
existing = await self.get(show_all=True)
await self.call_hq(
"sdk_RemoveVendor",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"vendor_name",
vendor_name,
)
]),
)
this_vendor_address_id = None
vendor_address_id_map = {}
for vendor in existing:
va_id = vendor.findtext("VendorAddressID")
if vendor.findtext("VendorName") == vendor_name:
this_vendor_address_id = va_id
vendor_address_id_map.setdefault(va_id, 0)
vendor_address_id_map[va_id] += 1
if vendor_address_id_map.get(this_vendor_address_id) == 1:
vendor_address_obj = VendorAddress(self.logger, self.hq_credentials)
await vendor_address_obj.delete(this_vendor_address_id)
[docs]
async def update(
self,
vendor_id,
vendor_name=DEFAULT,
parameters_bit_flag=DEFAULT,
audit_action_id=DEFAULT,
property_id=DEFAULT,
payload_stored_proc=DEFAULT,
purge_account_list=DEFAULT,
balance_hade_system_property=None,
serialize_for_cli=False,
):
"""Updates Q2_Vendors. Any value that is not passed takes the existing value in the db as default value"""
vendor_obj = Vendor(self.logger, self.hq_credentials, ret_table_obj=True)
vendor_list = await vendor_obj.get(vendor_id=vendor_id, show_all=True)
vendor = vendor_list[0]
system_property_data_obj = SystemPropertyData(
self.logger, self.hq_credentials, ret_table_obj=True
)
system_property_data_row = await system_property_data_obj.get(
name=balance_hade_system_property, strict=True
)
if balance_hade_system_property and not system_property_data_row:
raise DatabaseDataError(
f'No matching row with BALANCE_HADE_SYSTEM_PROPERTY: "{balance_hade_system_property}"'
)
parameters = ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.Int,
name="vendor_id",
value=vendor_id,
),
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.VarChar,
name="vendor_name",
value=vendor.VendorName if vendor_name == DEFAULT else vendor_name,
),
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.Int,
name="parameters_bit_flag",
value=vendor.ParametersBitflag
if parameters_bit_flag == DEFAULT
else parameters_bit_flag,
),
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.Int,
name="audit_action_id",
value=vendor.ActionID
if audit_action_id == DEFAULT
else audit_action_id,
),
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.Int,
name="property_id",
value=vendor.PropertyID if property_id == DEFAULT else property_id,
),
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.VarChar,
name="payload_stored_proc",
value=vendor.PayloadStoredProc
if payload_stored_proc == DEFAULT
else payload_stored_proc,
),
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.Bit,
name="purge_account_list",
value=vendor.PurgeAcctListOnSuccess
if purge_account_list == DEFAULT
else purge_account_list,
),
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.VarChar,
name="balance_hade_system_property",
value=balance_hade_system_property
if system_property_data_row
else None,
),
])
response = await self.call_hq("sdk_UpdateVendor", parameters)
if serialize_for_cli:
response = self.serialize_for_cli(
response,
[
"VendorID",
"VendorName",
"ActionID",
"ParametersBitflag",
"VendorAddressID",
"PropertyID",
"PayloadStoredProc",
],
)
return response