from typing import List, Optional
from argparse import _SubParsersAction
from datetime import datetime
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
from q2_sdk.hq.models.hq_params.stored_procedure import Param
from q2_sdk.hq.hq_api.wedge_online_banking import (
DisclaimerFormMarkAsAccepted as ob_DisclaimerFormMarkAsAccepted,
)
from q2_sdk.hq.hq_api.q2_api import (
DisclaimerFormMarkAsAccepted as api_DisclaimerFormMarkAsAccepted,
)
from q2_sdk.tools.decorators import dev_only
from ..db.user import User
from ..db.customer import Customer
from ..db.ui_text import UiText
from .db_object import DbObject
from .representation_row_base import RepresentationRowBase
D_TYPES = ExecuteStoredProcedure.DataType
[docs]
class DisclaimerRow(RepresentationRowBase):
DisclaimerID: IntElement = "DisclaimerID"
ShortName: StringElement = "ShortName"
DisplayName: StringElement = "DisplayName"
BundleDisplayName: StringElement = "BundleDisplayName"
DisclaimerElementID: IntElement = "DisclaimerElementID"
DisclaimerDataID: IntElement = "DisclaimerDataID"
DisclaimerFormDataID: IntElement = "DisclaimerFormDataID"
GroupID: IntElement = "GroupID"
[docs]
class Disclaimer(DbObject):
REPRESENTATION_ROW_CLASS = DisclaimerRow
GET_BY_NAME_KEY = "ShortName"
NAME = "Disclaimer"
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_disclaimers")
subparser.set_defaults(parser="get_disclaimers")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser = parser.add_parser("get_disclaimer_form_data")
subparser.set_defaults(parser="get_disclaimer_form_data")
subparser.set_defaults(
func=partial(self.get_disclaimer_form_data, serialize_for_cli=True)
)
group = subparser.add_mutually_exclusive_group()
group.add_argument(
"-f", "--form_id", type=int, help="Limit results to a single form"
)
group.add_argument(
"-e", "--short_name", help="Limit results to a single form by name"
)
subparser = parser.add_parser("add_disclaimer")
subparser.set_defaults(parser="add_disclaimer")
subparser.set_defaults(func=partial(self.create))
subparser.add_argument("display_name", help="Q2_DisclaimerBundle.DisplayName")
subparser.add_argument("corresponding_form_shortname", help="Q2_Form.ShortName")
subparser.add_argument(
"ui_text_elem_shortname", help="Q2_UiTextElement.ShortName"
)
subparser = parser.add_parser("remove_disclaimer")
subparser.set_defaults(parser="remove_disclaimer")
subparser.set_defaults(func=partial(self.delete))
subparser.add_argument("short_name", help="Q2_Disclaimer.ShortName")
subparser.add_argument(
"disclaimer_bundle_prefix", help="DisplayName Prefix for DisclaimerBundle"
)
subparser = parser.add_parser("accept_disclaimer_for_user_id")
subparser.set_defaults(parser="accept_disclaimer_for_user_id")
subparser.set_defaults(func=partial(self.accept_for_user_id))
subparser.add_argument("user_id", type=int, help="Q2_User.UserID")
subparser.add_argument(
"disclaimer_shortname_list", nargs="+", help="Q2_Disclaimer.ShortName"
)
[docs]
async def get(self, serialize_for_cli=False) -> List[DisclaimerRow]:
response = await self.call_hq("sdk_GetDisclaimers")
if serialize_for_cli:
response = self.serialize_for_cli(
response,
[
"DisclaimerID",
"DisclaimerBundleID",
"ShortName",
"DisplayName",
"BundleDisplayName",
"DisclaimerElementID",
"DisclaimerDataID",
"DisclaimerFormDataID",
"GroupID",
],
)
return response
[docs]
async def create(
self,
display_name,
corresponding_form_shortname,
ui_text_elem_shortname: Optional[str] = None,
short_name="",
disclaimer_type="",
):
if not short_name:
short_name = corresponding_form_shortname
if not ui_text_elem_shortname:
ui_text_elem_shortname = short_name
await UiText(self.logger, self.hq_credentials).create(
ui_text_elem_shortname, f"Disclaimer: {short_name}", ui_text_elem_shortname
)
params = self._build_create_parameters(
display_name,
corresponding_form_shortname,
ui_text_elem_shortname,
short_name,
disclaimer_type,
)
await self.call_hq(
"sdk_AddDisclaimer", ExecuteStoredProcedure.SqlParameters(params)
)
def _build_create_parameters(
self,
display_name,
corresponding_form_shortname,
ui_text_elem_shortname,
short_name,
disclaimer_type,
):
parameters = []
possible_params = [
Param(display_name, D_TYPES.VarChar, "display_name"),
Param(
corresponding_form_shortname,
D_TYPES.VarChar,
"corresponding_form_shortname",
),
Param(
ui_text_elem_shortname, D_TYPES.VarChar, "ui_text_elem_shortname", True
),
Param(short_name, D_TYPES.VarChar, "short_name"),
Param(disclaimer_type, D_TYPES.VarChar, "disclaimer_type"),
]
for item in possible_params:
item.add_to_param_list(parameters)
return parameters
[docs]
@dev_only
async def delete(self, short_name, disclaimer_bundle_prefix=None):
"""Note: this only works in the dev environment"""
if not disclaimer_bundle_prefix:
disclaimer_bundle_prefix = short_name
params = []
Param(short_name, D_TYPES.VarChar, "short_name").add_to_param_list(params)
Param(
disclaimer_bundle_prefix, D_TYPES.VarChar, "disclaimer_bundle_prefix"
).add_to_param_list(params)
return await self.call_hq(
"sdk_RemoveDisclaimer", ExecuteStoredProcedure.SqlParameters(params)
)
[docs]
async def accept_for_user_id(
self, user_id: int, disclaimer_shortname_list: List[str]
):
"""
:param user_id: Corresponds to Q2_User.UserID
:param disclaimer_shortname_list: List of Q2_Disclaimer.ShortName
"""
user = (
await User(
self.logger, hq_credentials=self.hq_credentials, ret_table_obj=True
).get(user_id=user_id)
)[0]
customer = (
await Customer(
self.logger, hq_credentials=self.hq_credentials, ret_table_obj=True
).get(customer_id=user.CustomerID.text)
)[0]
disclaimer_list = await self.get()
filtered_disclaimers = [
x
for x in disclaimer_list
if x.find("GroupID")
and x.GroupID.pyval == customer.GroupID
and x.ShortName.text in disclaimer_shortname_list
]
filtered_disclaimers_shortnames = [
x.ShortName.text for x in filtered_disclaimers
]
err = False
for name in disclaimer_shortname_list:
if name not in filtered_disclaimers_shortnames:
err = True
self.logger.error("Disclaimer not found for given user_id: %s", name)
if err:
raise DatabaseDataError
for disclaimer in filtered_disclaimers:
if disclaimer.findtext("DisclaimerDataID"):
await self._accept_non_form_disclaimer(user_id, disclaimer)
elif disclaimer.findtext("DisclaimerFormDataID"):
await self._accept_form_disclaimer(user_id, disclaimer)
self.logger.info(
"Disclaimer %s accepted for user_id %s",
disclaimer.ShortName.text,
user_id,
)
async def _accept_non_form_disclaimer(
self, user_id: int, disclaimer: DisclaimerRow
):
params = []
Param(user_id, D_TYPES.Int, "UserID").add_to_param_list(params)
Param(
disclaimer.DisclaimerDataID.pyval, D_TYPES.Int, "DisclaimerDataID"
).add_to_param_list(params)
Param(datetime.now(), D_TYPES.DateTime, "AcceptanceDate").add_to_param_list(
params
)
params_obj = ExecuteStoredProcedure.ParamsObj(
self.logger,
"Q2_DisclaimerAccept",
ExecuteStoredProcedure.SqlParameters(params),
hq_credentials=self.hq_credentials,
)
await ExecuteStoredProcedure.execute(params_obj)
async def _accept_form_disclaimer(self, user_id: int, disclaimer: DisclaimerRow):
if self.hq_credentials and self.hq_credentials.auth_token:
hq_endpoint = ob_DisclaimerFormMarkAsAccepted
else:
hq_endpoint = api_DisclaimerFormMarkAsAccepted
params_obj = hq_endpoint.ParamsObj(
self.logger,
user_id=user_id,
disclaimer_form_data_id=disclaimer.DisclaimerFormDataID.pyval,
hq_credentials=self.hq_credentials,
)
await hq_endpoint.execute(params_obj)