from argparse import _SubParsersAction
from functools import partial
from typing import Optional
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.hq.models.hq_params.stored_procedure import Param
from q2_sdk.hq.table_row import TableRow
from .db_object import DbObject
D_TYPES = ExecuteStoredProcedure.DataType
[docs]
class ThirdPartyDataRow(TableRow):
VendorID: int
UserID: int
DataID: int
DataValue: str
[docs]
class ThirdPartyData(DbObject):
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_third_party_data")
subparser.set_defaults(parser="get_third_party_data")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser.add_argument("user_id", help="Q2_User.UserID")
subparser.add_argument("vendor_name", help="Q2_Vendors.VendorName")
subparser = parser.add_parser("add_third_party_data")
subparser.set_defaults(parser="add_third_party_data")
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")
subparser = parser.add_parser("add_or_update_third_party_data")
subparser.set_defaults(parser="add_or_update_third_party_data")
subparser.set_defaults(func=partial(self.add_or_update))
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_name: str, serialize_for_cli=False):
response = await self.call_hq(
"sdk_GetThirdPartyData",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "userID", user_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"vendorName",
vendor_name,
),
]),
)
if serialize_for_cli:
response = self.serialize_for_cli(
response, ["VendorID", "UserID", "DataID", "DataValue", "VendorName"]
)
return response
[docs]
async def create(self, user_id, vendor_name, data_name, data_value):
return await self.call_hq(
"sdk_AddThirdPartyData",
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 add_or_update(self, user_id, vendor_name, data_name, data_value):
"""
Upserts third-party data, factoring in Q2_ThirdPartyDataShared and Q2_VendorGroupMembers.
"""
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(
"sdk_UpdateThirdPartyData",
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_RemoveThirdPartyData",
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
),
]),
)
[docs]
async def delete_all_for_vendor(self, vendor_name: str):
return await self.call_hq(
"sdk_RemoveAllVendorThirdPartyData",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"vendorName",
vendor_name,
),
]),
)
[docs]
async def get_by_vendor_paginated(
self,
vendor_id: Optional[int] = None,
vendor_name: Optional[str] = None,
page_number: int = 1,
return_count: int = 500,
) -> list[ThirdPartyDataRow]:
assert any([vendor_id, vendor_name]), (
"Either vendor_id or vendor_name need to be included"
)
offset = (page_number - 1) * return_count
assert offset >= 0, "page_number must be 1 or greater"
sql_params = []
Param(offset, D_TYPES.Int, "offset").add_to_param_list(sql_params)
Param(return_count, D_TYPES.Int, "returnCount").add_to_param_list(sql_params)
if vendor_id:
Param(vendor_id, D_TYPES.Int, "vendorID").add_to_param_list(sql_params)
if vendor_name:
Param(vendor_name, D_TYPES.VarChar, "vendorName").add_to_param_list(
sql_params
)
return await self.call_hq(
"sdk_GetThirdPartyDataByVendorPaginated",
ExecuteStoredProcedure.SqlParameters(sql_params),
)