from argparse import _SubParsersAction
from enum import StrEnum
from functools import partial
from lxml.objectify import IntElement, StringElement
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.core.exceptions import DatabaseDataError, HqResponseError
from q2_sdk.hq.models.hq_params.stored_procedure import Param
from q2_sdk.tools.decorators import dev_only
from q2_sdk.tools.utils import EnumContainsMetaClass
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase
D_TYPES = ExecuteStoredProcedure.DataType
[docs]
class GTFlavorRow(RepresentationRowBase):
GtFlavorID: IntElement = "GtFlavorID"
ShortName: StringElement = "ShortName"
Description: StringElement = "Description"
TransactionTypeID: IntElement = "TransactionTypeID"
TransactionType: StringElement = "TransactionType"
[docs]
class BaseGT(StrEnum, metaclass=EnumContainsMetaClass):
FundsTransfer = "FundsTransfer"
ExternalTransfer = "ExternalTransfer"
[docs]
class GtFlavor(DbObject):
GET_BY_NAME_KEY = "ShortName"
Name = "GTFlavor"
REPRESENTATION_ROW_CLASS = GTFlavorRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_gtflavors")
subparser.set_defaults(parser="get_gtflavors")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
[docs]
@dev_only
async def create(
self, flavor_name: str, flavor_description: str, base_gt: BaseGT | str
):
if isinstance(base_gt, str):
if base_gt not in BaseGT:
raise DatabaseDataError(f"{base_gt} is not supported for GT Flavors")
base_gt = BaseGT(base_gt)
base_gt = base_gt.value
params = []
Param(flavor_name, D_TYPES.VarChar, "GtFlavorShortName").add_to_param_list(
params
)
Param(
flavor_description, D_TYPES.VarChar, "GtFlavorDescription"
).add_to_param_list(params)
Param(base_gt, D_TYPES.VarChar, "BaseGtTypeShortName").add_to_param_list(params)
Param(True, D_TYPES.Bit, "CreateAuditActions").add_to_param_list(params)
try:
response = await self.call_hq(
"dba_CreateGtFlavor", ExecuteStoredProcedure.SqlParameters(params)
)
return response
except HqResponseError as err:
if "already exists" in err.args[0]:
return True
else:
raise
[docs]
async def get(self, serialize_for_cli: bool = False):
response = await self.call_hq("sdk_GetGTFlavors")
if serialize_for_cli:
columns = ["GtFlavorID", "ShortName", "Description", "TransactionType"]
response = self.serialize_for_cli(response, columns)
return response