from argparse import _SubParsersAction
from functools import partial
from typing import List
from lxml.objectify import IntElement, StringElement
from q2_sdk.hq.models.online_user import OnlineUser
from q2_sdk.hq.models.online_session import OnlineSession
from q2_sdk.hq.db.audit_record import AuditRecord
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase
[docs]
class UiConfigPropertyDataRow(RepresentationRowBase):
ConfigPropertyID: IntElement = "ConfigPropertyID"
PropertyName: StringElement = "PropertyName"
UiSource: StringElement = "UiSource"
PropertyDataType: StringElement = "PropertyDataType"
PropertyValue: StringElement = "PropertyValue"
[docs]
class UiConfigPropertyData(DbObject):
GET_BY_NAME_KEY = "PropertyName"
NAME = "UiConfigPropertyData"
REPRESENTATION_ROW_CLASS = UiConfigPropertyDataRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_ui_config_property_data")
subparser.add_argument(
dest="prefix",
help="Shortname prefix for Q2_UIConfigPropertyDataElements.PropertyName",
)
subparser.set_defaults(parser="get_ui_config_property_data")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser = parser.add_parser("add_ui_config_property_data")
subparser.set_defaults(parser="add_ui_config_property_data")
subparser.set_defaults(func=partial(self.create))
subparser.add_argument(
"property_name", help="Q2_UIConfigPropertyDataElements.PropertyName"
)
subparser.add_argument(
"property_data_type", help="Q2_UIConfigPropertyDataElement.PropertyDataType"
)
subparser.add_argument(
"property_value", help="Q2_UIConfigPropertyData.PropertyValue"
)
subparser = parser.add_parser("remove_ui_config_property_data")
subparser.set_defaults(parser="remove_ui_config_property_data")
subparser.set_defaults(func=partial(self.delete))
subparser.add_argument(
"property_name", help="Q2_UIConfigPropertyDataElements.PropertyName"
)
[docs]
async def get(
self, prefix, serialize_for_cli=False
) -> List[UiConfigPropertyDataRow]:
response = await self.call_hq(
"sdk_GetUIConfigPropertyData",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar, "prefix", prefix
)
]),
)
if serialize_for_cli:
response = self.serialize_for_cli(
response,
[
"ConfigPropertyID",
"PropertyName",
"UiSource",
"PropertyDataType",
"PropertyValue",
],
)
return response
[docs]
async def create(
self, property_name, property_data_type, property_value, is_central=False
):
ui_source_short_name = None
if is_central:
ui_source_short_name = "BackOffice"
return await self.call_hq(
"sdk_AddUIConfigPropertyData",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"property_name",
property_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"property_data_type",
property_data_type,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"property_value",
property_value,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"ui_source_short_name",
ui_source_short_name,
),
]),
)
[docs]
async def delete(self, property_name):
return await self.call_hq(
"sdk_RemoveUIConfigPropertyData",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"property_name",
property_name,
)
]),
)
[docs]
async def update(
self,
property_name: str,
property_value: str,
online_session: OnlineSession,
online_user: OnlineUser,
ui_source: str = None,
config_property_id: int = None,
):
if property_name and not config_property_id:
assert ui_source, (
"UI Source needs to be included if updating by property name"
)
result = await self.call_hq(
"sdk_UpdateUIConfigPropertyData",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"property_name",
property_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"ui_source",
ui_source,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"property_value",
property_value,
),
]),
)
else:
assert config_property_id, (
"Conflicting arguments. Please update by property name/ui source or property id"
)
result = await self.call_hq(
"sdk_UpdateUIConfigPropertyData",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"config_property_id",
config_property_id,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"property_name",
property_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"property_value",
property_value,
),
]),
)
if self.hq_response.success:
audit = AuditRecord(self.logger, self.hq_credentials)
details = f"UIConfigPropertyData updated. property_name: {property_name} ui_source: {ui_source}"
details += f"config_property_id: {config_property_id} property_value: {property_value}"
await audit.create(
details,
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