import json
from argparse import _SubParsersAction
from functools import partial
from typing import List, Optional
from lxml.objectify import IntElement, StringElement, BoolElement
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.models.installers.form_params import InstallParams as OnlineInstallParams
from q2_sdk.models.installers.central_params import (
InstallParams as CentralInstallParams,
)
from q2_sdk.models.installers.caliper_api_custom_params import (
InstallParams as CustomAPIInstallParams,
)
from q2_sdk.models.installers.ardent_params import InstallParams as ArdentInstallParams
from q2_sdk.models.installers.audit_params import InstallParams as AuditInstallParams
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase
from ...core.q2_logging.logger import Q2LoggerType
from ...tools.utils import calculate_url
InstallParams = (
OnlineInstallParams
| CentralInstallParams
| CustomAPIInstallParams
| ArdentInstallParams
| AuditInstallParams
)
[docs]
class WedgeAddressRow(RepresentationRowBase):
WedgeAddressID: IntElement = "WedgeAddressID"
WedgeAddressTypeName: StringElement = "WedgeAddressTypeName"
Url: StringElement = "Url"
VendorId: IntElement = "VendorId"
TimeoutInMilliSeconds: IntElement = "TimeoutInMilliSeconds"
Version: StringElement = "Version"
Config: StringElement = "Config"
WedgePayloadStoredProc: StringElement = "WedgePayloadStoredProc"
HqAuthTokenTypeID: IntElement = "HqAuthTokenTypeID"
HqAuthTokenType: StringElement = "HqAuthTokenType"
AuthTokenLifeInMinutes: IntElement = "AuthTokenLifeInMinutes"
ConnectTimeoutInMilli: IntElement = "ConnectTimeoutInMilli"
SendAccountList: BoolElement = "SendAccountList"
SendAccountDetails: IntElement = "SendAccountDetails"
AccountRightsBitFlag: IntElement = "AccountRightsBitFlag"
AccountRightsIsExplicit: BoolElement = "AccountRightsIsExplicit"
AccountHydraProductCodes: StringElement = "AccountHydraProductCodes"
SkipGetUnsignedToken: BoolElement = "SkipGetUnsignedToken"
[docs]
class WedgeAddressWithAuditActionsRow(RepresentationRowBase):
WedgeAddressID: IntElement = "WedgeAddressID"
WedgeAddressTypeName: StringElement = "WedgeAddressTypeName"
Url: StringElement = "Url"
VendorId: IntElement = "VendorId"
WedgePayloadStoredProc: StringElement = "WedgePayloadStoredProc"
SendAccountList: BoolElement = "SendAccountList"
SendAccountDetails: IntElement = "SendAccountDetails"
AuditAction: StringElement = "AuditAction"
HasAfterAuditAction: BoolElement = "HasAfterAuditAction"
HasBeforeAuditAction: BoolElement = "HasBeforeAuditAction"
[docs]
class WedgeAddress(DbObject):
GET_BY_NAME_KEY = "WedgeAddressTypeName"
NAME = "WedgeAddress"
REPRESENTATION_ROW_CLASS = WedgeAddressRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_wedge_addresses")
subparser.set_defaults(parser="get_wedge_addresses")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser.add_argument(
"-s", "--short-name", help="Q2_WedgeAddress.WedgeAddressTypeName"
)
subparser = parser.add_parser("get_wedge_addresses_tied_to_audits")
subparser.set_defaults(parser="get_wedge_addresses_tied_to_audits")
subparser.set_defaults(
func=partial(self.get_wa_tied_to_audit_actions, serialize_for_cli=True)
)
subparser = parser.add_parser("remove_wedge_address")
subparser.set_defaults(parser="remove_wedge_address")
subparser.set_defaults(func=partial(self.delete))
subparser.add_argument(
"short_name", help="Q2_WedgeAddress.WedgeAddressTypeName"
)
subparser = parser.add_parser("update_wedge_address")
subparser.set_defaults(parser="update_wedge_address")
subparser.set_defaults(func=partial(self.update, serialize_for_cli=True))
subparser.add_argument(
"wedge_address_type_name", help="Q2_WedgeAddress.WedgeAddressTypeName"
)
subparser.add_argument("-u", "--url", help="Url")
subparser.add_argument(
"-t", "--time-out-in-milli-seconds", help="TimeoutInMilliSeconds"
)
subparser.add_argument(
"-a", "--account-rights-bit-flag", help="AccountRightsBitFlag"
)
subparser.add_argument(
"-hq", "--hq-auth-token-type-id", help="HqAuthTokenTypeID"
)
subparser.add_argument("-c", "--config", help="Config")
[docs]
async def create(self, install_params: InstallParams):
return await self.call_hq(
"sdk_AddWedgeAddress",
_get_wedge_address_parameters(install_params, self.logger),
)
[docs]
async def get(
self, short_name: Optional[str] = None, serialize_for_cli=False
) -> List[WedgeAddressRow]:
sql_parameters = []
if short_name:
sql_parameters = [
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar, "short_name", short_name
)
]
response = await self.call_hq(
"sdk_GetWedgeAddresses",
ExecuteStoredProcedure.SqlParameters(sql_parameters),
)
if serialize_for_cli:
columns = [
"WedgeAddressID",
"WedgeAddressTypeName",
"Url",
"VendorId",
"Version",
"WedgePayloadStoredProc",
"SendAccountList",
"SendAccountDetails",
"HqAuthTokenTypeID",
"HqAuthTokenType",
"AuthTokenLifeInMinutes",
]
response = self.serialize_for_cli(response, columns)
return response
[docs]
async def get_wa_tied_to_audit_actions(
self, serialize_for_cli=False
) -> List[WedgeAddressWithAuditActionsRow]:
sql_parameters = []
response = await self.call_hq(
"sdk_GetWedgeAddressesTiedToAuditActions",
ExecuteStoredProcedure.SqlParameters(sql_parameters),
)
if serialize_for_cli:
columns = [
"WedgeAddressID",
"WedgeAddressTypeName",
"Url",
"VendorId",
"WedgePayloadStoredProc",
"SendAccountList",
"SendAccountDetails",
"AuditAction",
"HasAfterAuditAction",
"HasBeforeAuditAction",
]
response = self.serialize_for_cli(response, columns)
return response
[docs]
async def delete(self, short_name):
return await self.call_hq(
"sdk_RemoveWedgeAddress",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"short_name",
short_name,
)
]),
)
[docs]
async def update(
self,
wedge_address_type_name,
url=None,
time_out_in_milli_seconds=None,
account_rights_bit_flag=None,
hq_auth_token_type_id=None,
config=None,
auth_token_life_in_minutes=None,
skip_unsigned_token=True,
serialize_for_cli=False,
) -> list[WedgeAddressRow]:
wedge_address = WedgeAddress(
self.logger, self.hq_credentials, ret_table_obj=True
)
wedge_address_rows = await wedge_address.get(short_name=wedge_address_type_name)
wedge_address_row = wedge_address_rows[0]
wa_config = wedge_address_row.Config
if config:
config = config if isinstance(config, str) else json.dumps(config)
elif wa_config:
config = wa_config if isinstance(wa_config, str) else json.dumps(wa_config)
else:
config = ""
sql_parameters = ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.VarChar,
name="WedgeAddressTypeName",
value=wedge_address_type_name,
),
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.VarChar,
name="Url",
value=wedge_address_row.Url if not url else url,
),
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.Int,
name="TimeoutInMilliSeconds",
value=wedge_address_row.TimeoutInMilliSeconds
if not time_out_in_milli_seconds
else time_out_in_milli_seconds,
),
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.Int,
name="AccountRightsBitFlag",
value=wedge_address_row.AccountRightsBitFlag
if not account_rights_bit_flag
else account_rights_bit_flag,
),
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.Int,
name="HqAuthTokenTypeID",
value=wedge_address_row.HqAuthTokenTypeID
if not hq_auth_token_type_id
else hq_auth_token_type_id,
),
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.VarChar,
name="Config",
value=config,
),
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.Int,
name="AuthTokenLifeInMinutes",
value=wedge_address_row.AuthTokenLifeInMinutes
if not auth_token_life_in_minutes
else auth_token_life_in_minutes,
),
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.Bit,
name="SkipGetUnsignedToken",
value=wedge_address_row.SkipGetUnsignedToken
if not skip_unsigned_token
else skip_unsigned_token,
),
])
response = await self.call_hq("sdk_UpdateWedgeAddress", sql_parameters)
if serialize_for_cli:
response = self.serialize_for_cli(
response,
[
"WedgeAddressID",
"WedgeAddressTypeName",
"Url",
"VendorId",
"TimeoutInMilliSeconds",
"Config",
"HqAuthTokenTypeID",
"AccountRightsBitFlag",
],
)
return response
def _get_wedge_address_parameters(
install_params: InstallParams, logger: Q2LoggerType | None
):
if not hasattr(install_params, "address_target"):
install_params.address_target = calculate_url(vars(install_params))
if logger:
logger.warning("InstallParams object was missing 'address_target'")
if not hasattr(install_params, "config_json"):
install_params.config_json = json.dumps(install_params.db_plan)
if logger:
logger.warning("InstallParams object was missing 'config_json'")
if not hasattr(install_params, "time_out_in_milli_seconds"):
install_params.time_out_in_milli_seconds = None
if logger:
logger.warning(
"InstallParams object was missing 'time_out_in_milli_seconds'"
)
if not hasattr(install_params, "hq_auth_token_type"):
install_params.hq_auth_token_type = None
if logger:
logger.warning("InstallParams object was missing 'hq_auth_token_type'")
if not hasattr(install_params, "auth_token_life_in_minutes"):
install_params.auth_token_life_in_minutes = None
if logger:
logger.warning(
"InstallParams object was missing 'auth_token_life_in_minutes'"
)
sql_parameters = ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"wedge_address_type_name",
install_params.short_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"url",
install_params.address_target,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"config",
install_params.config_json,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"account_rights_bit_flags",
install_params.account_rights_bit_flags,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Bit,
"account_rights_is_explicit",
install_params.account_rights_is_explicit,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"account_hydra_product_codes",
install_params.account_hydra_product_codes,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Bit,
"send_account_list",
install_params.send_account_list,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"account_details_enum",
install_params.send_account_details,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"wedge_payload_stored_proc",
install_params.wedge_payload_stored_proc,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"time_out_in_milli_seconds",
install_params.time_out_in_milli_seconds,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"hq_auth_token_type",
install_params.hq_auth_token_type,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"auth_token_life_in_minutes",
install_params.auth_token_life_in_minutes,
),
])
return sql_parameters