from argparse import _SubParsersAction
from functools import partial
from typing import Optional
from lxml.objectify import IntElement, StringElement
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.tools.decorators import dev_only
from q2_sdk.hq.models.hq_params.stored_procedure import Param
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase
D_TYPES = ExecuteStoredProcedure.DataType
[docs]
class AdminUserPropertyDataRow(RepresentationRowBase):
UserPropertyDataID: IntElement = "UserPropertyDataID"
UISourceID: IntElement = "UISourceID"
GroupID: IntElement = "GroupID"
UserID: IntElement = "UserID"
PropertyID: IntElement = "PropertyID"
PropertyValue: StringElement = "PropertyValue"
PropertyLongName: StringElement = "PropertyLongName"
PropertyName: StringElement = "PropertyName"
PropertyDataType: StringElement = "PropertyDataType"
Weight: IntElement = "Weight"
IsGroupProperty: bool = "IsGroupProperty"
IsUserProperty: bool = "IsUserProperty"
[docs]
class AdminGroupPropertyRow(RepresentationRowBase):
UserPropertyDataID: IntElement = "UserPropertyDataID"
GroupID: IntElement = "GroupID"
PropertyID: IntElement = "PropertyID"
PropertyValue: StringElement = "PropertyValue"
PropertyName: StringElement = "PropertyName"
PropertyLongName: StringElement = "PropertyLongName"
[docs]
class AdminUserPropertyData(DbObject):
REPRESENTATION_ROW_CLASS = AdminUserPropertyDataRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_admin_user_property_data")
subparser.set_defaults(parser="get_admin_user_property_data")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser.add_argument(
"property_name",
help="Q2_AdminUserPropertyDataElements.PropertyValue prefix",
)
subparser.add_argument(
"-u", "--user-id", help="Q2_AdminUserPropertyData.UserID"
)
subparser.add_argument(
"-g", "--group-id", help="Q2_AdminUserPropertyData.GroupID"
)
subparser.add_argument(
"--strict",
action="store_true",
help="Only show exact matches for group/user",
)
subparser.add_argument(
"--return-count",
type=int,
help="Max number of matches to return",
default=10,
)
subparser.add_argument("--ui-source", help="Q2_UISource.ShortName")
subparser.add_argument(
"--no-trunc", action="store_true", help="Do not truncate PropertyLongName"
)
subparser = parser.add_parser("add_admin_user_property_data")
subparser.set_defaults(parser="add_admin_user_property_data")
subparser.set_defaults(func=partial(self.create))
subparser.add_argument(
"property_name", help="Q2_AdminUserPropertyDataElement.PropertyName"
)
subparser.add_argument(
"property_value", help="Q2_AdminUserPropertyData.PropertyValue"
)
subparser.add_argument(
"-u", "--user-id", help="Q2_AdminUserPropertyData.UserID"
)
subparser.add_argument(
"-g", "--group-id", help="Q2_AdminUserPropertyData.GroupID"
)
subparser.add_argument("-f", "--fi-id", help="Q2_AdminUserPropertyData.FIID")
subparser.add_argument("--ui-source", help="Q2_UISource.ShortName")
subparser = parser.add_parser("update_admin_user_property_data")
subparser.set_defaults(parser="update_admin_user_property_data")
subparser.set_defaults(func=partial(self.update))
subparser.add_argument(
"property_name", help="Q2_AdminUserPropertyDataElement.PropertyName"
)
subparser.add_argument(
"property_value", help="Q2_AdminUserPropertyData.PropertyValue"
)
subparser.add_argument(
"-u", "--user-id", help="Q2_AdminUserPropertyData.UserID"
)
subparser.add_argument(
"-g", "--group-id", help="Q2_AdminUserPropertyData.GroupID"
)
subparser.add_argument("-f", "--fi-id", help="Q2_AdminUserPropertyData.FIID")
subparser.add_argument("--ui-source", help="Q2_UISource.ShortName")
subparser = parser.add_parser("update_admin_user_property_data_by_id")
subparser.set_defaults(parser="update_admin_user_property_data_by_id")
subparser.set_defaults(func=partial(self.update_by_id))
subparser.add_argument(
"admin_user_property_data_id",
help="Q2_AdminUserPropertyData.UserPropertyDataID",
)
subparser.add_argument(
"property_name", help="Q2_AdminUserPropertyDataElement.PropertyName"
)
subparser.add_argument(
"property_value", help="Q2_AdminUserPropertyData.PropertyValue"
)
subparser = parser.add_parser("remove_admin_user_property_data")
subparser.set_defaults(parser="remove_admin_user_property_data")
subparser.set_defaults(func=partial(self.delete))
subparser.add_argument(
"property_name", help="Q2_AdminUserPropertyDataElement.PropertyName"
)
subparser.add_argument(
"-u", "--user-id", help="Q2_AdminUserPropertyData.UserID"
)
subparser.add_argument(
"-g", "--group-id", help="Q2_AdminUserPropertyData.GroupID"
)
subparser.add_argument("-f", "--fi-id", help="Q2_AdminUserPropertyData.FIID")
subparser.add_argument("--ui-source", help="Q2_UISource.ShortName")
subparser = parser.add_parser("get_admin_user_property_data_by_value")
subparser.set_defaults(parser="get_admin_user_property_data_by_value")
subparser.set_defaults(func=partial(self.get_by_value, serialize_for_cli=True))
subparser.add_argument(
"property_value", help="Q2_AdminUserPropertyData.PropertyValue"
)
subparser.add_argument(
"admin_user_property_name",
help="Q2_AdminUserPropertyDataElements.PropertyName",
)
subparser.add_argument(
"--no-trunc", action="store_true", help="Do not truncate PropertyLongName"
)
subparser = parser.add_parser("get_admin_group_property_data")
subparser.set_defaults(parser="get_admin_group_property_data")
subparser.set_defaults(
func=partial(self.get_all_group_properties, serialize_for_cli=True)
)
subparser.add_argument(
"-g", "--target-group-id", help="Q2_AdminUserPropertyData.GroupID"
)
subparser.add_argument(
"--exclude-default-value-rows",
action="store_false",
help="Only show exact matches for group",
dest="include_default_value_rows",
)
subparser.add_argument(
"--page-number",
type=int,
help="page number for pagination",
default=1,
dest="page_number",
)
subparser.add_argument(
"--page-size",
type=int,
help="number of rows per page",
dest="page_size",
default=200,
)
[docs]
async def get(
self,
property_name: str,
user_id: Optional[int] = None,
group_id: Optional[int] = None,
fi_id: Optional[int] = None,
ui_source: Optional[str] = None,
return_count=10,
strict=True,
no_trunc=False,
serialize_for_cli=False,
) -> list[AdminUserPropertyDataRow]:
truncate = not no_trunc
response = await self.call_hq(
"sdk_GetAdminUserPropertyData",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"PropertyName",
property_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "UserID", user_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "GroupID", group_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "FIID", fi_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar, "UISource", ui_source
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"ReturnCount",
str(return_count),
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Bit, "Strict", str(strict)
),
]),
)
if serialize_for_cli:
fields_to_truncate = []
if truncate:
fields_to_truncate = ["PropertyValue"]
response = self.serialize_for_cli(
response,
fields_to_display=[
"PropertyID",
"PropertyName",
"UserID",
"GroupID",
"FIID",
"PropertyValue",
],
fields_to_truncate=fields_to_truncate,
)
return response
[docs]
async def get_by_value(
self,
property_value: str,
admin_user_property_name: str,
no_trunc=False,
serialize_for_cli=False,
) -> list[AdminUserPropertyDataRow]:
truncate = not no_trunc
response = await self.call_hq(
"sdk_GetAdminUserPropertyDataByValue",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"property_value",
property_value,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"admin_user_property_name",
admin_user_property_name,
),
]),
)
if serialize_for_cli:
fields_to_truncate = []
if truncate:
fields_to_truncate = ["TextValue"]
response = self.serialize_for_cli(
response,
fields_to_display=[
"PropertyID",
"UserID",
"GroupID",
"FIID",
"PropertyValue",
],
fields_to_truncate=fields_to_truncate,
)
return response
[docs]
async def get_all_group_properties(
self,
target_group_id: Optional[int] = None,
include_default_value_rows: bool = True,
page_number: int = 1,
page_size: int = 200,
serialize_for_cli=True,
) -> list[AdminGroupPropertyRow]:
"""
Gets values from the Q2_AdminUserPropertyData table for properties marked as group properties
:param target_group_id: Optional parameter to return only properties for a specific group
:param include_default_value_rows: default value rows hold property values that are used if a group does not have a row for a property. These rows will not have GroupID values returned (NULL in the database)
:param page_number: the starting point for pagination.
:param page_size: the number of rows to return per page
:param serialize_for_cli: flag for calling from the cli
"""
offset = (page_number - 1) * page_size
assert offset >= 0, "page_number must be 1 or greater"
sql_params = []
Param(target_group_id, D_TYPES.Int, "targetGroupID").add_to_param_list(
sql_params
)
Param(
include_default_value_rows, D_TYPES.Bit, "includeDefaultValue"
).add_to_param_list(sql_params)
Param(offset, D_TYPES.Int, "offsetRows").add_to_param_list(sql_params)
Param(page_size, D_TYPES.Int, "returnCount").add_to_param_list(sql_params)
response = await self.call_hq(
"sdk_GetAdminGroupProperties",
ExecuteStoredProcedure.SqlParameters(sql_params),
)
if serialize_for_cli:
fields_to_truncate = []
response = self.serialize_for_cli(
response,
fields_to_display=[
"PropertyID",
"PropertyName",
"GroupID",
"PropertyValue",
],
fields_to_truncate=fields_to_truncate,
)
return response
[docs]
async def create(
self,
property_name: str,
property_value: str,
user_id: Optional[int] = None,
group_id: Optional[int] = None,
fi_id: Optional[int] = None,
ui_source: Optional[str] = None,
):
return await self.call_hq(
"sdk_AddAdminUserPropertyData",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "UserID", user_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"PropertyName",
property_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"PropertyValue",
property_value,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar, "UISource", ui_source
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "GroupID", group_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "FIID", fi_id
),
]),
)
[docs]
async def update(
self,
property_name: str,
property_value: str,
user_id: Optional[int] = None,
group_id: Optional[int] = None,
fi_id: Optional[int] = None,
ui_source: Optional[str] = None,
):
return await self.call_hq(
"sdk_UpdateAdminUserPropertyData",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "UserID", user_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"PropertyName",
property_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"PropertyValue",
property_value,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar, "UISource", ui_source
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "GroupID", group_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "FIID", fi_id
),
]),
)
[docs]
async def update_by_id(
self, admin_user_property_data_id: int, property_name: str, property_value: str
):
assert int(admin_user_property_data_id) >= 1, (
"User Property Data ID must be an integer greater than 0"
)
return await self.call_hq(
"sdk_UpdateAdminUserPropertyDataById",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"UserPropertyDataId",
admin_user_property_data_id,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"PropertyName",
property_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"PropertyValue",
property_value,
),
]),
)
[docs]
@dev_only
async def delete(
self,
property_name: str,
user_id: Optional[int] = None,
group_id: Optional[int] = None,
fi_id: Optional[int] = None,
ui_source: Optional[str] = None,
strict=True,
):
"""Note: this only works in the dev environment"""
return await self.call_hq(
"sdk_RemoveAdminUserPropertyData",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "UserID", user_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"PropertyName",
property_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar, "UISource", ui_source
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "GroupID", group_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "FIID", fi_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Bit, "Strict", str(strict)
),
]),
)