from argparse import _SubParsersAction
import asyncio
from functools import partial
from lxml.objectify import IntElement, StringElement, BoolElement
from q2_sdk.hq.models.hq_params.stored_procedure import Param
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.core.exceptions import DatabaseDataError, DbObjectError
from q2_sdk.core.cli import textui
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase
from .user_property_data import UserPropertyData
D_TYPES = ExecuteStoredProcedure.DataType
[docs]
class UserPropertyDataElementRow(RepresentationRowBase):
PropertyID: IntElement = "PropertyID"
PropertyName: StringElement = "PropertyName"
PropertyLongName: StringElement = "PropertyLongName"
PropertyDataType: StringElement = "PropertyDataType"
UserPropertyCategoryID: IntElement = "UserPropertyCategoryID"
AllowUserEdit: BoolElement = "AllowUserEdit"
AllowUserView: BoolElement = "AllowUserView"
AllowCustView: BoolElement = "AllowCustView"
[docs]
class UserPropertyDataElement(DbObject):
REPRESENTATION_ROW_CLASS = UserPropertyDataElementRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_user_property_data_elements")
subparser.set_defaults(parser="get_user_property_data_elements")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser.add_argument(
"property_name_prefix",
help="Wildcard match against Q2_UserPropertyDataElements.PropertyName",
)
subparser.add_argument(
"--no-trunc", action="store_true", help="Do not truncate PropertyLongName"
)
subparser = parser.add_parser("add_user_property_data_element")
subparser.set_defaults(parser="add_user_property_data_element")
subparser.set_defaults(func=partial(self.create))
subparser.add_argument(
"property_name", help="Q2_UserPropertyDataElements.PropertyName"
)
subparser.add_argument(
"property_long_name", help="Q2_UserPropertyDataElements.PropertyLongName"
)
subparser.add_argument("data_type", help="BOO, STR, BIT, INT, DBL")
subparser.add_argument(
"--category", help="Q2_UserPropertyDataCategory.ShortName", default="Custom"
)
subparser.add_argument(
"--allow_user_edit",
default=False,
action="store_true",
help="Q2_UserPropertyDataElements.AllowUserEdit",
)
subparser.add_argument(
"--allow_user_view",
default=False,
action="store_true",
help="Q2_UserPropertyDataElements.AllowUserView",
)
subparser.add_argument(
"--allow_cust_view",
default=False,
action="store_true",
help="Q2_UserPropertyDataElements.AllowCustView",
)
subparser = parser.add_parser("update_user_property_data_element")
subparser.set_defaults(parser="update_user_property_data_element")
subparser.set_defaults(func=partial(self.update, serialize_for_cli=True))
subparser.add_argument(
"property_name", help="Q2_UserPropertyDataElements.PropertyName"
)
subparser.add_argument(
"property_long_name", help="Q2_UserPropertyDataElements.PropertyLongName"
)
subparser = parser.add_parser("remove_user_property_data_element")
subparser.set_defaults(parser="remove_user_property_data_element")
subparser.set_defaults(func=partial(self.delete, interactive=True))
subparser.add_argument(
"property_name", help="Q2_UserPropertyDataElement.ShortName"
)
subparser = parser.add_parser("get_user_property_data_elements_by_category")
subparser.description = "Returns all user property data elements that matches with the given category shortname"
subparser.set_defaults(parser="get_user_property_data_elements_by_category")
subparser.set_defaults(
func=partial(self.get_by_category_name, serialize_for_cli=True)
)
subparser.add_argument(
"user_property_category_short_name",
type=str,
help="Q2_UserPropertyCategory.ShortName",
)
def _serialize_response(self, overrride_fields: list, truncate: bool, response):
fields_to_truncate = []
if truncate:
fields_to_truncate = ["PropertyLongName"]
response = self.serialize_for_cli(
response,
fields_to_display=[
"PropertyID",
"PropertyName",
"PropertyLongName",
"PropertyDataType",
*overrride_fields,
],
fields_to_truncate=fields_to_truncate,
)
return response
[docs]
async def get(
self, property_name_prefix: str, no_trunc=False, serialize_for_cli=False
) -> list[UserPropertyDataElementRow]:
truncate = not no_trunc
params = []
Param(
property_name_prefix, D_TYPES.VarChar, "property_name_prefix"
).add_to_param_list(params)
response = await self.call_hq(
"sdk_GetUserPropertyDataElements",
ExecuteStoredProcedure.SqlParameters(params),
)
if serialize_for_cli:
response = self._serialize_response(
[
"UserPropertyCategoryID",
"AllowUserEdit",
"AllowUserView",
"AllowCustView",
],
truncate,
response,
)
return response
[docs]
async def get_by_id(self, user_property_element_id: int):
params = []
Param(user_property_element_id, D_TYPES.Int, "property_id").add_to_param_list(
params
)
return await self.call_hq(
"sdk_GetUserPropertyDataElementsById",
ExecuteStoredProcedure.SqlParameters(params),
)
[docs]
async def get_one(self, property_name_prefix: str, strict=False):
user_property_data_elements = await self.get(property_name_prefix)
if user_property_data_elements != []:
if not strict:
return user_property_data_elements[0]
for element in user_property_data_elements:
if property_name_prefix == str(element.PropertyName):
return user_property_data_elements[0]
raise DatabaseDataError(f'No matching row with prefix "{property_name_prefix}"')
[docs]
async def create(
self,
property_name,
property_long_name,
data_type="STR",
category="Custom",
allow_user_edit=False,
allow_user_view=False,
allow_cust_view=False,
):
sql_params = []
list_of_params = [
Param(property_name, D_TYPES.VarChar, "property_name"),
Param(property_long_name, D_TYPES.VarChar, "property_long_name"),
Param(data_type, D_TYPES.VarChar, "property_data_type"),
Param(category, D_TYPES.VarChar, "category"),
Param(str(allow_user_edit), D_TYPES.Bit, "allow_user_edit"),
Param(str(allow_user_view), D_TYPES.Bit, "allow_user_view"),
Param(str(allow_cust_view), D_TYPES.Bit, "allow_cust_view"),
]
for item in list_of_params:
item.add_to_param_list(sql_params)
response = await self.call_hq(
"sdk_AddUserPropertyDataElement",
ExecuteStoredProcedure.SqlParameters(sql_params),
)
return response
[docs]
async def update(self, property_name, property_long_name, serialize_for_cli=False):
sql_params = []
Param(property_name, D_TYPES.VarChar, "property_name").add_to_param_list(
sql_params
)
Param(
property_long_name, D_TYPES.VarChar, "property_long_name"
).add_to_param_list(sql_params)
response = await self.call_hq(
"sdk_UpdateUserPropertyDataElement",
ExecuteStoredProcedure.SqlParameters(sql_params),
)
if serialize_for_cli:
response = self.serialize_for_cli(response)
return response
[docs]
async def get_by_category_name(
self,
user_property_category_short_name: str,
no_trunc=False,
serialize_for_cli=False,
):
truncate = not no_trunc
sql_params = []
Param(
user_property_category_short_name,
D_TYPES.VarChar,
"user_property_category_short_name",
).add_to_param_list(sql_params)
response = await self.call_hq(
"sdk_GetUserPropertyDataElementsByCategoryShortName",
ExecuteStoredProcedure.SqlParameters(sql_params),
)
if serialize_for_cli:
fields_to_truncate = []
if truncate:
fields_to_truncate = ["PropertyLongName"]
response = self.serialize_for_cli(
response,
fields_to_display=[
"PropertyID",
"PropertyName",
"PropertyLongName",
"PropertyDataType",
"UserPropertyCategoryID",
"AllowUserEdit",
"AllowUserView",
"AllowCustView",
],
fields_to_truncate=fields_to_truncate,
)
return response
[docs]
async def delete(self, property_name, interactive=False):
"""Note: this only works in the dev environment"""
if interactive:
upd_obj = UserPropertyData(
self.logger, hq_credentials=self.hq_credentials, ret_table_obj=True
)
attached_properties = await upd_obj.get(property_name, strict=False)
confirmed = True
if attached_properties:
confirmed = textui.query_yn(
"You are about to remove %d properties followed by the parent element. Ok?"
% len(attached_properties)
)
if confirmed:
if attached_properties:
deletions = []
for elem in attached_properties:
deletions.append(
upd_obj.delete(
elem.PropertyName,
user_id=elem.UserID,
group_id=elem.GroupID,
customer_id=elem.CustomerID,
host_account_data=elem.HostAccountID,
ui_source=elem.UISourceID,
user_role_id=elem.UserRoleID,
)
)
await asyncio.gather(*deletions)
else:
raise DbObjectError("Aborted")
sql_params = []
Param(property_name, D_TYPES.VarChar, "property_name").add_to_param_list(
sql_params
)
return await self.call_hq(
"sdk_RemoveUserPropertyDataElement",
ExecuteStoredProcedure.SqlParameters(sql_params),
)