from argparse import _SubParsersAction
from enum import Enum
from functools import partial
from typing import List
from lxml.objectify import BoolElement, IntElement, StringElement
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.hq.db.host_tran_code_group import HostTranCodeGroup
from q2_sdk.hq.db.ofx_transaction_type import OfxTransactionType
from q2_sdk.tools.decorators import dev_only
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase
[docs]
class HostTranCodeRow(RepresentationRowBase):
HostTranCodeID: IntElement = "HostTranCodeID"
HostTranCode: StringElement = "HostTranCode"
TranCodeGroupID: IntElement = "TranCodeGroupID"
DorC: StringElement = "DorC"
TranCodeDesc: StringElement = "TranCodeDesc"
ShowImage: BoolElement = "ShowImage"
VoiceFileId: StringElement = "VoiceFileId"
TransactionType: StringElement = "TransactionType"
OfxTrnType: StringElement = "OfxTrnType"
OfxCheckNumHandling: StringElement = "OfxCheckNumHandling"
ShowTransactionItemDetail: StringElement = "ShowTransactionItemDetail"
BaiTransactionTypeID: StringElement = "BaiTransactionTypeID"
[docs]
class DOrC(Enum):
Deposit = "D"
Credit = "C"
[docs]
class HostTranCode(DbObject):
NAME = "HostTranCode"
GET_BY_NAME_KEY = "HostTranCode"
REPRESENTATION_ROW_CLASS = HostTranCodeRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_host_tran_codes")
subparser.set_defaults(parser="get_host_tran_codes")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser.add_argument(
"--no-trunc", action="store_true", help="Do not truncate TranCodeDesc"
)
subparser = parser.add_parser("add_host_tran_code")
subparser.set_defaults(parser="add_host_tran_code")
subparser.set_defaults(func=partial(self.create))
subparser.add_argument("host_tran_code")
subparser.add_argument("tran_code_group_name", type=str)
subparser.add_argument("d_or_c", help="D or C")
subparser.add_argument("tran_code_desc")
subparser.add_argument("ofx_transaction_type_name")
subparser.add_argument("--show_image", action="store_true")
subparser.add_argument("--transaction_type")
subparser.add_argument("--ofx_check_num_handling")
subparser.add_argument("--show_transaction_item_detail")
subparser = parser.add_parser("remove_host_tran_code")
subparser.set_defaults(parser="remove_host_tran_code")
subparser.set_defaults(func=partial(self.delete))
subparser.add_argument(
"host_tran_code_name", help="Q2_HostTranCode.HostTranCode"
)
[docs]
async def get(
self, no_trunc=False, serialize_for_cli=False
) -> List[HostTranCodeRow]:
truncate = not no_trunc
response = await self.call_hq("sdk_GetHostTranCode")
if serialize_for_cli:
columns = [
"HostTranCodeID",
"HostTranCode",
"TranCodeGroupID",
"DorC",
"TranCodeDesc",
]
fields_to_truncate = []
if truncate:
fields_to_truncate = ["TranCodeDesc"]
response = self.serialize_for_cli(
response,
fields_to_display=columns,
fields_to_truncate=fields_to_truncate,
)
return response
[docs]
async def create(
self,
host_tran_code: str,
tran_code_group_name: str,
d_or_c: DOrC,
tran_code_desc: str,
ofx_transaction_type_name: str,
show_image=False,
transaction_type=0,
ofx_check_num_handling=0,
show_transaction_item_detail=0,
):
if isinstance(d_or_c, DOrC):
d_or_c = d_or_c.value
htcg_obj = HostTranCodeGroup(self.logger, self.hq_credentials)
htcg = await htcg_obj.get_by_name(tran_code_group_name)
tran_code_group_id = htcg.TranCodeGroupID.pyval
ott_obj = OfxTransactionType(self.logger, self.hq_credentials)
ott = await ott_obj.get_by_name(ofx_transaction_type_name)
ofx_transaction_type_id = ott.TransactionTypeID.pyval
parameters = [
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"host_tran_code",
host_tran_code,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"tran_code_group_id",
tran_code_group_id,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Char, "d_or_c", d_or_c
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"tran_code_desc",
tran_code_desc,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"ofx_transaction_type_id",
ofx_transaction_type_id,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Bit, "show_image", show_image
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"transaction_type",
transaction_type,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"ofx_check_num_handling",
ofx_check_num_handling,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"show_transaction_item_detail",
show_transaction_item_detail,
),
]
await self.call_hq(
"sdk_AddHostTranCode", ExecuteStoredProcedure.SqlParameters(parameters)
)
[docs]
@dev_only
async def delete(self, host_tran_code_name):
"""Note: this only works in the dev environment"""
await self.call_hq(
"sdk_RemoveHostTranCode",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"host_tran_code_name",
host_tran_code_name,
)
]),
)