from argparse import _SubParsersAction
from functools import partial
from typing import List
from lxml.objectify import IntElement, StringElement
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from .db_object import DbObject, DEFAULT
from .representation_row_base import RepresentationRowBase
[docs]
class ThirdPartyDataSharedRow(RepresentationRowBase):
VendorID: IntElement = "VendorID"
UserID: IntElement = "UserID"
GroupID: IntElement = "GroupID"
DataID: IntElement = "DataID"
DataValue: StringElement = "DataValue"
Name: StringElement = "Name"
Description: StringElement = "Description"
[docs]
class ThirdPartyDataShared(DbObject):
NAME = "ThirdPartyDataShared"
REPRESENTATION_ROW_CLASS = ThirdPartyDataSharedRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_third_party_data_shared")
subparser.set_defaults(parser="get_third_party_data_shared")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser.add_argument("user_id", help="Q2_User.UserID")
subparser.add_argument("group_id", help="Q2_Group.GroupID")
subparser.add_argument("vendor_id", help="Q2_Vendors.VendorID")
subparser = parser.add_parser("add_third_party_data_shared")
subparser.set_defaults(parser="add_third_party_data_shared")
subparser.set_defaults(func=partial(self.create))
subparser.add_argument("user_id", help="Q2_User.UserID")
subparser.add_argument("vendor_name", help="Q2_Vendors.VendorName")
subparser.add_argument("data_name", help="Q2_ThirdPartyDataElements.Name")
subparser.add_argument("data_value", help="Q2_ThirdPartyData.DataValue")
[docs]
async def get(
self,
user_id: int,
vendor_id: int,
group_id: int = DEFAULT,
serialize_for_cli=False,
) -> List[ThirdPartyDataSharedRow]:
if group_id == DEFAULT:
group_id = -1
response = await self.call_hq(
"Q2_GetThirdPartyData",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "UserID", user_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "VendorID", vendor_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "GroupID", group_id
),
]),
)
if serialize_for_cli:
response = self.serialize_for_cli(
response,
[
"VendorID",
"UserID",
"GroupID",
"DataID",
"DataValue",
"Name",
"Description",
],
)
return response
[docs]
async def create(self, user_id, vendor_name, data_name, data_value):
return await self.call_hq(
"Q2_AddUpdateThirdPartyData",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "UserID", user_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"VendorName",
vendor_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar, "DataName", data_name
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar, "DataValue", data_value
),
]),
)
[docs]
async def update(self, user_id, vendor_name, data_name, data_value):
return await self.call_hq(
"Q2_AddUpdateThirdPartyData",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "UserID", user_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"VendorName",
vendor_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar, "DataName", data_name
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar, "DataValue", data_value
),
]),
)
[docs]
async def delete(self, user_id: int, vendor_name: str, data_name: str):
return await self.call_hq(
"sdk_RemoveThirdPartyDataShared",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "userID", user_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"vendorName",
vendor_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar, "dataName", data_name
),
]),
)