import json
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.models.installers.sso_params import InstallParams
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase
[docs]
class VendorAddressRow(RepresentationRowBase):
VendorAddressID: IntElement = "VendorAddressID"
Address: StringElement = "Address"
Description: StringElement = "Description"
TimeoutInSeconds: IntElement = "TimeoutInSeconds"
[docs]
class VendorAddress(DbObject):
NAME = "VendorAddress"
REPRESENTATION_ROW_CLASS = VendorAddressRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_vendor_address")
subparser.set_defaults(parser="get")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser = parser.add_parser("update_vendor_address")
subparser.set_defaults(parser="update")
subparser.set_defaults(func=partial(self.update))
subparser.add_argument("vendor_name", help="Q2_Vendors.VendorName")
subparser.add_argument("address", help="Q2_VendorAddress.Address")
subparser = parser.add_parser("remove_vendor_address")
subparser.set_defaults(parser="delete")
subparser.set_defaults(func=partial(self.delete))
subparser.add_argument(
"vendor_address_id", help="Q2_VendorAddress.VendorAddressID"
)
[docs]
async def get(self, serialize_for_cli=False) -> list[VendorAddressRow]:
response = await self.call_hq("sdk_GetVendorAddress")
if serialize_for_cli:
columns = ["VendorAddressID", "Address", "Description"]
response = self.serialize_for_cli(response, columns)
return response
[docs]
async def create(self, address: str, description: str) -> VendorAddressRow:
assert isinstance(address, str), "Please supply a valid address"
assert isinstance(description, str), "Please supply a valid description"
response = await self.call_hq(
"sdk_CreateVendorAddress",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar, "address", address
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"description",
description,
),
]),
)
return response[0]
[docs]
async def update(self, vendor_name: str, address: str):
response = await self.call_hq(
"sdk_UpdateVendorAddress",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"vendor_name",
vendor_name,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar, "address", address
),
]),
)
return response
[docs]
async def delete(self, vendor_address_id: int):
response = await self.call_hq(
"sdk_RemoveVendorAddress",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "id", vendor_address_id
)
]),
)
return response
[docs]
async def delete_by_url(self, address: str):
assert isinstance(address, str), "Please supply a valid address"
all_current = await self.get()
existing: Optional[VendorAddressRow] = None
for row in all_current:
if row.Address.text == address:
existing = row
response = None
if existing is not None:
response = await self.call_hq(
"sdk_RemoveVendorAddress",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"id",
existing.VendorAddressID.pyval,
)
]),
)
return response
[docs]
async def create_from_parameters_object(self, install_params: InstallParams):
address_target = install_params.get_address(
base_url_override=install_params.install_base_url,
prefix=install_params.prefix,
)
if not install_params.description:
author = {"sdk": "Q2"}
if install_params.is_customer_created:
author = {"sdk": "Customer", "company": install_params.company}
description = json.dumps(author)
else:
description = install_params.description
return (
await self.create(address_target, description),
address_target,
description,
)