from argparse import _SubParsersAction
from functools import partial
from enum import Enum
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
from .representation_row_base import RepresentationRowBase
from lxml.etree import tostring
from lxml.objectify import E, IntElement, StringElement, BoolElement
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 .db_object import DbObject
D_TYPES = ExecuteStoredProcedure.DataType
[docs]
class EStatementDateColumn(Enum):
CreateDate = "Create"
LastChange = "Last"
[docs]
@dataclass
class AddOrUpdateParams:
"""
:param host_account_id: The account id of the target account
:param opt_in: boolean representing opting in or out of e-statements
:param email_address: the target email address
:param user_id: The user ID making the creation or update
:param alt_email_address: An alternate email for e-statements
:params last_change: The timestamp of the last change. Defaults to datetime.now
"""
host_account_id: int
opt_in: bool
email_address: str
user_id: int
alt_email_address: Optional[str] = None
last_change: Optional[datetime] = None
[docs]
def build_sql_parameters(self):
if not self.last_change:
self.last_change = datetime.now()
params = []
for param in [
Param(self.host_account_id, D_TYPES.Int, "HostAccountID"),
Param(self.opt_in, D_TYPES.Bit, "OptIn"),
Param(self.email_address, D_TYPES.VarChar, "EmailAddress"),
Param(self.user_id, D_TYPES.Int, "ModifiedByID"),
Param(self.last_change.isoformat(), D_TYPES.DateTime, "LastChange"),
]:
param.add_to_param_list(params)
if self.alt_email_address:
Param(
self.alt_email_address, D_TYPES.VarChar, "AlternateEmailAddress"
).add_to_param_list(params)
# The stored proc has required but unused parameters. Passing those values here
Param("NotUsed", D_TYPES.VarChar, "AccountNumberInternal").add_to_param_list(
params
)
Param("NotUsed", D_TYPES.VarChar, "AccountNickName").add_to_param_list(params)
return params
[docs]
class EStatementRow(RepresentationRowBase):
HostAccountID: IntElement = "HostAccountID"
AccountNumberInternal: StringElement = "AccountNumberInternal"
AccountNumberExternal: StringElement = "AccountNumberExternal"
CreateDate: StringElement = "CreateDate"
LastChange: StringElement = "LastChange"
ModifiedByID: IntElement = "ModifiedByID"
ModifiedFirstName: StringElement = "ModifiedFirstName"
ModifiedLastName: StringElement = "ModifiedLastName"
OptIn: BoolElement = "OptIn"
HasBeenReported: IntElement = "HasBeenReported"
EmailID: IntElement = "EmailID"
EmailAddress: StringElement = "EmailAddress"
AlternateEmailID: IntElement = "AlternateEmailID"
AlternateEmailAddress: StringElement = "AlternateEmailAddress"
ProcessedDate: StringElement = "ProcessedDate"
[docs]
class EStatement(DbObject):
"""
Queries the Q2_Customer table for rows that match the given arguments. Gathers additional details on specific
customers
"""
NAME = "EStatement"
REPRESENTATION_ROW_CLASS = EStatementRow
[docs]
def add_arguments(self, parser: _SubParsersAction):
subparser = parser.add_parser("get_e_statement")
subparser.set_defaults(parser="get")
subparser.set_defaults(func=partial(self.get, serialize_for_cli=True))
subparser.add_argument(
"host_account_ids",
nargs="*",
help="Q2_HostAccount.HostAccountID. "
"If multiple are passed in, ids must be delimited by a space",
)
[docs]
async def get(
self, host_account_ids: list[int] | str, serialize_for_cli=False
) -> list[EStatementRow]:
host_account_ids = [int(id) for id in host_account_ids]
assert isinstance(host_account_ids, list), (
"host_account_ids parameter must be a list of integers"
)
# <request"><get ha="1" /><get ha="2" /><get ha="3" /></request>
get_ids = [E.get(ha=str(id)) for id in host_account_ids]
request = tostring(E.request(*get_ids), encoding="utf-8")
params = []
Param(request, D_TYPES.Xml, "request").add_to_param_list(params)
response = await self.call_hq(
"sdk_GetEStatement",
ExecuteStoredProcedure.SqlParameters(params),
use_json=False,
)
if serialize_for_cli:
columns = [
"HostAccountID",
"CreateDate",
"LastChange",
"ModifiedByID",
"OptIn",
"HasBeenReported",
"EmailID",
"AlternateEmailID",
"ProcessedDate",
]
response = self.serialize_for_cli(response, columns)
return response
[docs]
async def get_by_date_range(
self,
start_date: datetime,
end_date: datetime,
date_column: EStatementDateColumn,
page_number=1,
page_size=200,
) -> list[EStatementRow]:
"""
Get e-statements by date range and supports pagination
:param start_date: The start date of the date range, as a datetime object
:param end_date: The end date of the date range, as a datetime object
:param date_column: The date column to filter the date on
:param page_number: the starting point for pagination. Defaults to 1
:param page_size: The number of transactions to get per page
:return: A list of EStatement rows that fit the provided parameters
"""
offset = (page_number - 1) * page_size
assert offset >= 0, "page_number must be 1 or greater"
sql_params = []
Param(start_date.isoformat(), D_TYPES.DateTime, "FromDate").add_to_param_list(
sql_params
)
Param(end_date.isoformat(), D_TYPES.DateTime, "ToDate").add_to_param_list(
sql_params
)
Param(offset, D_TYPES.Int, "Offset").add_to_param_list(sql_params)
Param(page_size, D_TYPES.Int, "ReturnCount").add_to_param_list(sql_params)
match date_column:
case EStatementDateColumn.CreateDate:
stored_proc_name = "sdk_GetEStatementsByCreateDate"
case EStatementDateColumn.LastChange:
stored_proc_name = "sdk_GetEStatementsByLastChange"
case _:
raise DatabaseDataError("Date Column not supported")
response = await self.call_hq(
stored_proc_name, ExecuteStoredProcedure.SqlParameters(sql_params)
)
return response
[docs]
async def add_or_update(self, e_statement_values: AddOrUpdateParams):
sql_params = e_statement_values.build_sql_parameters()
return await self.call_hq(
"Q2_EStatementUpdate", ExecuteStoredProcedure.SqlParameters(sql_params)
)