from argparse import _SubParsersAction
from functools import partial
from typing import List
from lxml.objectify import IntElement, StringElement
from q2_sdk.core.cli.textui import colored, puts_err
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase
[docs]
class AdapterAddressRow(RepresentationRowBase):
AddressID: IntElement = "AdapterAddressID"
Address: StringElement = "Address"
Description: StringElement = "Description"
UserName: StringElement = "UserName"
Password: StringElement = "Password"
ProcessingTypeID: IntElement = "ProcessingTypeID"
[docs]
class AdapterAddress(DbObject):
NAME = "AdapterAddress"
GET_BY_NAME_KEY = "Address"
REPRESENTATION_ROW_CLASS = AdapterAddressRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_adapter_addresses")
subparser.set_defaults(parser="get")
subparser.add_argument("-i", "--address_id", help="Q2_AdapterAddress.AddressID")
subparser.add_argument(
"--no-filter",
dest="http_only",
action="store_false",
help="Show non-http addresses",
)
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser = parser.add_parser("add_adapter_address")
subparser.set_defaults(parser="add")
subparser.set_defaults(func=partial(self.create))
subparser.add_argument("address", help="Q2_AdapterAddresses.Address")
subparser.add_argument("description", help="Q2_AdapterAddresses.Description")
subparser.add_argument(
"processing_type_id", help="Q2_AdapterAddresses.ProcessingTypeID"
)
subparser = parser.add_parser("remove_adapter_address")
subparser.set_defaults(parser="delete")
subparser.set_defaults(func=partial(self.delete))
subparser.add_argument("address_id", help="Q2_AdapterAddress.AddressID")
subparser = parser.add_parser("update_adapter_address")
subparser.set_defaults(parser="update")
subparser.set_defaults(func=partial(self.update, serialize_for_cli=True))
subparser.add_argument("address_id", help="Q2_AdapterAddress.AddressID")
subparser.add_argument("-a", "--address", help="Q2_AdapterAddress.Address")
subparser.add_argument(
"-d", "--description", help="Q2_AdapterAddress.Description"
)
subparser.add_argument(
"-p", "--processing_type_id", help="Q2_AdapterAddress.ProcessingTypeID"
)
[docs]
async def get(
self,
address_id=None,
http_only=True,
address_value=None,
processing_type_id=None,
serialize_for_cli=False,
suppress_messages=False,
) -> List[AdapterAddressRow]:
"""
:param address_id: Q2_AdapterAddresses.AddressID
:param http_only: If True, will only show addresses prefixed with http
:param serialize_for_cli: Used when running from the command line
"""
response = await self.call_hq("sdk_GetAdapterAddresses")
if http_only:
response = [a for a in response if "http" in a.Address.text]
if address_id:
response = [a for a in response if int(address_id) == int(a.AddressID.text)]
if not response and not suppress_messages:
puts_err(colored.red(f"Adapter address ID {address_id} not found"))
if address_value:
response = [a for a in response if address_value == a.Address.text]
if not response and not suppress_messages:
puts_err(colored.red(f"Adapter address {address_value} not found"))
if processing_type_id:
response = [
a
for a in response
if int(processing_type_id) == int(a.ProcessingTypeID.text)
]
if not response and not suppress_messages:
puts_err(
colored.red(
f"Adapter address not found with processing type id {processing_type_id}"
)
)
if serialize_for_cli:
columns = ["AddressID", "Address", "Description", "ProcessingTypeID"]
response = self.serialize_for_cli(response, columns)
return response
[docs]
async def create(
self, address: str, description: str, processing_type_id: str
) -> AdapterAddressRow:
assert address and isinstance(address, str), "Please supply a valid address"
assert description and isinstance(description, str), (
"Please supply a valid description"
)
assert processing_type_id and isinstance(processing_type_id, str), (
"Please supply a valid ProcessingTypeId"
)
response = await self.call_hq(
"sdk_AddAdapterAddress",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar, "address", address
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"description",
description,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"processing_type_id",
processing_type_id,
),
]),
)
return response[0]
[docs]
async def delete(self, address_id: int):
assert address_id, "Please supply a valid address ID"
response = await self.call_hq(
"sdk_RemoveAdapterAddress",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "address_id", address_id
)
]),
)
return response
[docs]
async def update(
self,
address_id,
address=None,
description=None,
processing_type_id=None,
serialize_for_cli=False,
):
adapter_address_obj = AdapterAddress(self.logger, self.hq_credentials)
adapter_address_rows = await adapter_address_obj.get(address_id)
if not adapter_address_rows:
response = []
else:
adapter_address_row = adapter_address_rows[0]
sql_parameters = ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.Int,
name="address_id",
value=address_id,
),
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.VarChar,
name="address",
value=adapter_address_row.Address.text if not address else address,
),
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.VarChar,
name="description",
value=adapter_address_row.Description.text
if not description
else description,
),
ExecuteStoredProcedure.SqlParam(
data_type=ExecuteStoredProcedure.DataType.Int,
name="processing_type_id",
value=adapter_address_row.ProcessingTypeID.text
if not processing_type_id
else processing_type_id,
),
])
response = await self.call_hq("sdk_UpdateAdapterAddress", sql_parameters)
if serialize_for_cli:
response = self.serialize_for_cli(
response, ["AddressID", "Address", "Description", "ProcessingTypeID"]
)
return response