from argparse import _SubParsersAction
from functools import partial
from typing import Optional
from lxml.objectify import IntElement
from q2_sdk.core.dynamic_imports import (
api_ExecuteStoredProcedure as ExecuteStoredProcedure,
)
from q2_sdk.hq.table_row import TableRow, datetime
from .db_object import DbObject
from .disclaimer import Disclaimer
[docs]
class LogonDisclaimerAcceptanceRow(TableRow):
DisclaimerAcceptanceID: int
DisclaimerDataID: int
DisclaimerDataHistoryID: int
UserID: int
AcceptanceDate: datetime
[docs]
class LogonDisclaimerAcceptance(DbObject):
GET_BY_NAME_KEY = "UserID"
REPRESENTATION_ROW_CLASS = LogonDisclaimerAcceptanceRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_logon_disclaimer_acceptance")
subparser.set_defaults(parser="get")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser.add_argument("user_id", type=int, help="dbo.Q2_User.UserID")
subparser.add_argument("short_name", help="dbo.Q2_Disclaimer.ShortName")
[docs]
async def get(
self, user_id: int, short_name: str, serialize_for_cli=False
) -> list[LogonDisclaimerAcceptanceRow]:
assert isinstance(user_id, int) and user_id > 0, (
"user_id must be an integer > 0"
)
assert short_name and short_name.strip(), (
"short_name must be provided and not empty"
)
disclaimer_data_id = await self._get_disclaimer_data_id_by_short_name(
short_name
)
if disclaimer_data_id is None:
return []
response = await self.call_hq(
"sdk_GetLogonDisclaimerAcceptance",
ExecuteStoredProcedure.SqlParameters([
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int, "user_id", user_id
),
ExecuteStoredProcedure.SqlParam(
ExecuteStoredProcedure.DataType.Int,
"disclaimer_data_id",
disclaimer_data_id,
),
]),
)
if serialize_for_cli:
columns = ["AcceptanceDate"]
response = self.serialize_for_cli(response, columns)
return response
async def _get_disclaimer_data_id_by_short_name(
self, short_name: str
) -> Optional[IntElement]:
disclaimer_obj = Disclaimer(
self.logger, self.hq_credentials, ret_table_obj=True
)
disclaimers = await disclaimer_obj.get()
for disclaimer in disclaimers:
if disclaimer.ShortName.text == short_name:
return disclaimer.DisclaimerDataID
return None