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 .db_object import DbObject
from .representation_row_base import RepresentationRowBase
from ...tools.decorators import dev_only
[docs]
class EdvDetailRow(RepresentationRowBase):
# object name: type hinting = "column name in the db response"
EdvDetailID: IntElement = "EdvDetailID"
EdvProfileID: IntElement = "EdvProfileID"
AuditActionShortName: StringElement = "AuditActionShortName"
AuditActionID: IntElement = "AuditActionID"
[docs]
class EdvDetail(DbObject):
# GET_BY_NAME_KEY = "column in the db response"
NAME = "EdvDetail"
REPRESENTATION_ROW_CLASS = EdvDetailRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("add_edv_detail")
subparser.set_defaults(parser="add_edv_detail")
subparser.set_defaults(func=partial(self.add, serialize_for_cli=True))
subparser.add_argument(
"audit_action_shortname", help="Q2_AuditAction.ShortName"
)
subparser.add_argument("edv_profile_id", help="Q2_EdvProfile.EdvProfileID")
subparser = parser.add_parser("get_edv_detail")
subparser.set_defaults(parser="get_edv_detail")
subparser.set_defaults(
func=partial(self.get_by_edv_profile_id, serialize_for_cli=True)
)
subparser.add_argument("edv_profile_id", help="Q2_EdvProfile.EdvProfileID")
subparser = parser.add_parser("remove_edv_detail")
subparser.set_defaults(parser="remove_edv_detail")
subparser.set_defaults(
func=partial(
self.remove_edv_detail_by_edv_profile_id, serialize_for_cli=True
)
)
subparser.add_argument("edv_profile_id", help="Q2_EdvProfile.EdvProfileID")
subparser.add_argument(
"--audit_action_shortname", help="Q2_AuditAction.ShortName"
)
[docs]
async def add(
self,
audit_action_shortname: str,
edv_profile_id: int,
require_token_if_not_suspect=True,
):
await self.call_hq(
"sdk_AddEdvDetail",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"audit_action_shortname",
audit_action_shortname,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"edv_profile_id",
int(edv_profile_id),
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Bit,
"require_token_if_not_suspect",
require_token_if_not_suspect,
),
]),
)
[docs]
async def get_by_edv_profile_id(self, edv_profile_id: int, serialize_for_cli=False):
response = await self.call_hq(
"sdk_GetEdvDetailByEdvProfileID",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"edv_profile_id",
int(edv_profile_id),
)
]),
)
if serialize_for_cli:
columns = [
"EdvDetailID",
"EdvProfileID",
"AuditActionShortName",
"AuditActionID",
]
response = self.serialize_for_cli(response, columns)
return response
[docs]
@dev_only
async def remove_edv_detail_by_edv_profile_id(
self, edv_profile_id: int, audit_action_shortname: Optional[str] = None
):
await self.call_hq(
"sdk_RemoveEdvDetail",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.VarChar,
"audit_action_shortname",
audit_action_shortname,
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"edv_profile_id",
int(edv_profile_id),
),
]),
)