# This is an autogenerated file from the command "q2 generate_hq_api" and will be overwritten if run again
"""
DataSet models for AddCustomer.
These typed dataclass models represent the tables and columns within the
DataSet parameters for the AddCustomer HQ operation.
IMPORTANT: Update operations require a GET first to populate the original_record
field. HQ uses the unchanged row for optimistic concurrency checking.
Usage::
from q2_sdk.hq.hq_api.backoffice import AddCustomer
q2_customer = AddCustomer.Q2Customer(
customer_name="...",
is_company=True,
create_date="...",
auto_generated=True,
group_id=1,
charge_plan_start_date="...",
)
q2_phone_number = AddCustomer.Q2PhoneNumber(
country_id=1,
city_or_area_code="...",
local_number="...",
phone_type=1,
)
q2_customer_account = AddCustomer.Q2CustomerAccount(
customer_id=1,
access=1,
primary_only=True,
)
q2_customer_account_access_view = AddCustomer.Q2CustomerAccountAccessView(
customer_account_id=1,
customer_id=1,
current_access_flag=1,
)
q2_subsidiary_view = AddCustomer.Q2SubsidiaryView(
subsidiary_id=1,
customer_id=1,
display_name="...",
)
q2_address = AddCustomer.Q2Address(
street_address1="...",
city="...",
address_type=1,
country_id=1,
is_international=True,
)
transaction_rights = AddCustomer.TransactionRights(
transaction_type_id=1,
transaction_type="...",
description="...",
enabled=True,
authorize=True,
cancel=True,
create=True,
create_restricted=True,
update=True,
view=1,
limit_per_transaction=1.0,
limit_per_day=1.0,
limit_per_month=1.0,
limit_per_acct_per_day=1.0,
count_per_acct_per_day=1,
count_per_day=1,
count_per_month=1,
dual_approval_limit=1.0,
token_required_limit=1.0,
bit_flags=1,
)
user_rights = AddCustomer.UserRights(
rights_short_name="...",
rights_long_name="...",
has_rights=True,
)
user_monetary_rights = AddCustomer.UserMonetaryRights(
rights_short_name="...",
rights_long_name="...",
amount=1.0,
)
q2_customer_account_view = AddCustomer.Q2CustomerAccountView(
customer_account_id=1,
customer_id=1,
host_account_id=1,
account_number_internal="...",
product_name="...",
is_external_account=True,
product_id=1,
product_type_id=1,
user_access=1,
current_access_flag=1,
)
"""
from __future__ import annotations
import copy
from dataclasses import dataclass, field
from typing import Any, ClassVar, Optional
from q2_sdk.hq.models.dataset_builder import DataSetBuilder, DataSetTable, RowState
[docs]
@dataclass
class Q2Customer:
"""
Represents a row in the Q2_Customer table.
"""
customer_id: int = -1
customer_name: str = ""
tax_id: Optional[str] = None
is_company: bool = False
service_charge_plan_id: Optional[int] = None
charge_account: Optional[int] = None
create_date: str = ""
default_address_id: Optional[int] = None
auto_generated: bool = False
deleted_date: Optional[str] = None
group_id: int = 0
charge_plan_start_date: str = ""
cust_info: Optional[str] = None
primary_cif: Optional[str] = None
policy_id: Optional[int] = None
company_policy_id: Optional[int] = None
host_user: Optional[str] = None
host_pwd: Optional[str] = None
legal_entity_identifier: Optional[str] = None
host_account_not_allowed_for_ach: Optional[bool] = None
restricted_entitlement_profile_id: Optional[int] = None
restricted_entitlement_last_change: Optional[str] = None
original_record: Optional["Q2Customer"] = field(default=None, repr=False)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["CustomerID", "customer_id"],
["CustomerName", "customer_name"],
["TaxID", "tax_id"],
["IsCompany", "is_company"],
["ServiceChargePlanID", "service_charge_plan_id"],
["ChargeAccount", "charge_account"],
["CreateDate", "create_date"],
["DefaultAddressID", "default_address_id"],
["AutoGenerated", "auto_generated"],
["DeletedDate", "deleted_date"],
["GroupID", "group_id"],
["ChargePlanStartDate", "charge_plan_start_date"],
["CustInfo", "cust_info"],
["PrimaryCIF", "primary_cif"],
["PolicyID", "policy_id"],
["CompanyPolicyID", "company_policy_id"],
["HostUser", "host_user"],
["HostPwd", "host_pwd"],
["LegalEntityIdentifier", "legal_entity_identifier"],
["HostAccountNotAllowedForAch", "host_account_not_allowed_for_ach"],
["RestrictedEntitlementProfileID", "restricted_entitlement_profile_id"],
["RestrictedEntitlementLastChange", "restricted_entitlement_last_change"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.customer_id,
self.customer_name,
self.tax_id,
self.is_company,
self.service_charge_plan_id,
self.charge_account,
self.create_date,
self.default_address_id,
self.auto_generated,
self.deleted_date,
self.group_id,
self.charge_plan_start_date,
self.cust_info,
self.primary_cif,
self.policy_id,
self.company_policy_id,
self.host_user,
self.host_pwd,
self.legal_entity_identifier,
self.host_account_not_allowed_for_ach,
self.restricted_entitlement_profile_id,
self.restricted_entitlement_last_change,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"customer_id": int,
"customer_name": str,
"tax_id": str,
"is_company": bool,
"service_charge_plan_id": int,
"charge_account": int,
"create_date": str,
"default_address_id": int,
"auto_generated": bool,
"deleted_date": str,
"group_id": int,
"charge_plan_start_date": str,
"cust_info": str,
"primary_cif": str,
"policy_id": int,
"company_policy_id": int,
"host_user": str,
"host_pwd": str,
"legal_entity_identifier": str,
"host_account_not_allowed_for_ach": bool,
"restricted_entitlement_profile_id": int,
"restricted_entitlement_last_change": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2Customer":
"""Parse an lxml row element into this model with original_record set."""
kwargs: dict[str, Any] = {}
for hq_name, python_name in cls.get_columns():
child = row_elem.find(hq_name)
if child is not None and child.text is not None:
target_type = cls._COLUMN_TYPES.get(python_name, str)
if target_type is bool:
kwargs[python_name] = child.text.lower() in ("true", "1")
elif target_type in (int, float):
try:
kwargs[python_name] = target_type(child.text)
except (ValueError, TypeError):
kwargs[python_name] = child.text
else:
kwargs[python_name] = child.text
instance = cls(**kwargs)
instance.original_record = copy.deepcopy(instance)
return instance
[docs]
@dataclass
class Q2PhoneNumber:
"""
Represents a row in the Q2_PhoneNumber table.
"""
phone_id: int = -1
country_id: int = 0
city_or_area_code: str = ""
local_number: str = ""
extension: Optional[str] = None
phone_type: int = 0
user_id: Optional[int] = None
customer_id: Optional[int] = None
original_record: Optional["Q2PhoneNumber"] = field(default=None, repr=False)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["PhoneID", "phone_id"],
["CountryID", "country_id"],
["CityOrAreaCode", "city_or_area_code"],
["LocalNumber", "local_number"],
["Extension", "extension"],
["PhoneType", "phone_type"],
["UserID", "user_id"],
["CustomerID", "customer_id"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.phone_id,
self.country_id,
self.city_or_area_code,
self.local_number,
self.extension,
self.phone_type,
self.user_id,
self.customer_id,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"phone_id": int,
"country_id": int,
"city_or_area_code": str,
"local_number": str,
"extension": str,
"phone_type": int,
"user_id": int,
"customer_id": int,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2PhoneNumber":
"""Parse an lxml row element into this model with original_record set."""
kwargs: dict[str, Any] = {}
for hq_name, python_name in cls.get_columns():
child = row_elem.find(hq_name)
if child is not None and child.text is not None:
target_type = cls._COLUMN_TYPES.get(python_name, str)
if target_type is bool:
kwargs[python_name] = child.text.lower() in ("true", "1")
elif target_type in (int, float):
try:
kwargs[python_name] = target_type(child.text)
except (ValueError, TypeError):
kwargs[python_name] = child.text
else:
kwargs[python_name] = child.text
instance = cls(**kwargs)
instance.original_record = copy.deepcopy(instance)
return instance
[docs]
@dataclass
class Q2CustomerAccount:
"""
Represents a row in the Q2_CustomerAccount table.
"""
customer_account_id: int = -1
customer_id: int = 0
host_internal_cif: Optional[str] = None
host_account_id: Optional[int] = None
access: int = 0
primary_only: bool = False
original_record: Optional["Q2CustomerAccount"] = field(default=None, repr=False)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["CustomerAccountID", "customer_account_id"],
["CustomerID", "customer_id"],
["HostInternalCIF", "host_internal_cif"],
["HostAccountID", "host_account_id"],
["Access", "access"],
["PrimaryOnly", "primary_only"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.customer_account_id,
self.customer_id,
self.host_internal_cif,
self.host_account_id,
self.access,
self.primary_only,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"customer_account_id": int,
"customer_id": int,
"host_internal_cif": str,
"host_account_id": int,
"access": int,
"primary_only": bool,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2CustomerAccount":
"""Parse an lxml row element into this model with original_record set."""
kwargs: dict[str, Any] = {}
for hq_name, python_name in cls.get_columns():
child = row_elem.find(hq_name)
if child is not None and child.text is not None:
target_type = cls._COLUMN_TYPES.get(python_name, str)
if target_type is bool:
kwargs[python_name] = child.text.lower() in ("true", "1")
elif target_type in (int, float):
try:
kwargs[python_name] = target_type(child.text)
except (ValueError, TypeError):
kwargs[python_name] = child.text
else:
kwargs[python_name] = child.text
instance = cls(**kwargs)
instance.original_record = copy.deepcopy(instance)
return instance
[docs]
@dataclass
class Q2CustomerAccountAccessView:
"""
Represents a row in the Q2_CustomerAccountAccessView table.
"""
customer_account_id: int = 0
customer_id: int = 0
host_internal_cif: Optional[str] = None
host_account_id: Optional[int] = None
product_id: Optional[int] = None
product_type_id: Optional[int] = None
current_access_flag: int = 0
valid_system_access_flag: Optional[int] = None
original_record: Optional["Q2CustomerAccountAccessView"] = field(
default=None, repr=False
)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["CustomerAccountID", "customer_account_id"],
["CustomerID", "customer_id"],
["HostInternalCIF", "host_internal_cif"],
["HostAccountID", "host_account_id"],
["ProductID", "product_id"],
["ProductTypeID", "product_type_id"],
["CurrentAccessFlag", "current_access_flag"],
["ValidSystemAccessFlag", "valid_system_access_flag"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.customer_account_id,
self.customer_id,
self.host_internal_cif,
self.host_account_id,
self.product_id,
self.product_type_id,
self.current_access_flag,
self.valid_system_access_flag,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"customer_account_id": int,
"customer_id": int,
"host_internal_cif": str,
"host_account_id": int,
"product_id": int,
"product_type_id": int,
"current_access_flag": int,
"valid_system_access_flag": int,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2CustomerAccountAccessView":
"""Parse an lxml row element into this model with original_record set."""
kwargs: dict[str, Any] = {}
for hq_name, python_name in cls.get_columns():
child = row_elem.find(hq_name)
if child is not None and child.text is not None:
target_type = cls._COLUMN_TYPES.get(python_name, str)
if target_type is bool:
kwargs[python_name] = child.text.lower() in ("true", "1")
elif target_type in (int, float):
try:
kwargs[python_name] = target_type(child.text)
except (ValueError, TypeError):
kwargs[python_name] = child.text
else:
kwargs[python_name] = child.text
instance = cls(**kwargs)
instance.original_record = copy.deepcopy(instance)
return instance
[docs]
@dataclass
class Q2SubsidiaryView:
"""
Represents a row in the Q2_SubsidiaryView table.
"""
subsidiary_id: int = 0
customer_id: int = 0
display_name: str = ""
ach_name: Optional[str] = None
ach_tax_id: Optional[str] = None
wire_name: Optional[str] = None
wire_address1: Optional[str] = None
wire_address2: Optional[str] = None
wire_city: Optional[str] = None
wire_state: Optional[str] = None
wire_postal_code: Optional[str] = None
wire_country_id: Optional[int] = None
wire_is_international: Optional[bool] = None
wire_province: Optional[str] = None
wire_address_id: Optional[int] = None
original_record: Optional["Q2SubsidiaryView"] = field(default=None, repr=False)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["SubsidiaryID", "subsidiary_id"],
["CustomerID", "customer_id"],
["DisplayName", "display_name"],
["AchName", "ach_name"],
["AchTaxId", "ach_tax_id"],
["WireName", "wire_name"],
["WireAddress1", "wire_address1"],
["WireAddress2", "wire_address2"],
["WireCity", "wire_city"],
["WireState", "wire_state"],
["WirePostalCode", "wire_postal_code"],
["WireCountryID", "wire_country_id"],
["WireIsInternational", "wire_is_international"],
["WireProvince", "wire_province"],
["WireAddressID", "wire_address_id"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.subsidiary_id,
self.customer_id,
self.display_name,
self.ach_name,
self.ach_tax_id,
self.wire_name,
self.wire_address1,
self.wire_address2,
self.wire_city,
self.wire_state,
self.wire_postal_code,
self.wire_country_id,
self.wire_is_international,
self.wire_province,
self.wire_address_id,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"subsidiary_id": int,
"customer_id": int,
"display_name": str,
"ach_name": str,
"ach_tax_id": str,
"wire_name": str,
"wire_address1": str,
"wire_address2": str,
"wire_city": str,
"wire_state": str,
"wire_postal_code": str,
"wire_country_id": int,
"wire_is_international": bool,
"wire_province": str,
"wire_address_id": int,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2SubsidiaryView":
"""Parse an lxml row element into this model with original_record set."""
kwargs: dict[str, Any] = {}
for hq_name, python_name in cls.get_columns():
child = row_elem.find(hq_name)
if child is not None and child.text is not None:
target_type = cls._COLUMN_TYPES.get(python_name, str)
if target_type is bool:
kwargs[python_name] = child.text.lower() in ("true", "1")
elif target_type in (int, float):
try:
kwargs[python_name] = target_type(child.text)
except (ValueError, TypeError):
kwargs[python_name] = child.text
else:
kwargs[python_name] = child.text
instance = cls(**kwargs)
instance.original_record = copy.deepcopy(instance)
return instance
[docs]
@dataclass
class Q2Address:
"""
Represents a row in the Q2_Address table.
"""
address_id: int = -1
street_address1: str = ""
street_address2: Optional[str] = None
city: str = ""
state: Optional[str] = None
postal_code: Optional[str] = None
customer_id: Optional[int] = None
address_type: int = 0
user_id: Optional[int] = None
country_id: int = 0
is_international: bool = False
province: Optional[str] = None
original_record: Optional["Q2Address"] = field(default=None, repr=False)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["AddressID", "address_id"],
["StreetAddress1", "street_address1"],
["StreetAddress2", "street_address2"],
["City", "city"],
["State", "state"],
["PostalCode", "postal_code"],
["CustomerID", "customer_id"],
["AddressType", "address_type"],
["UserID", "user_id"],
["CountryID", "country_id"],
["IsInternational", "is_international"],
["Province", "province"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.address_id,
self.street_address1,
self.street_address2,
self.city,
self.state,
self.postal_code,
self.customer_id,
self.address_type,
self.user_id,
self.country_id,
self.is_international,
self.province,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"address_id": int,
"street_address1": str,
"street_address2": str,
"city": str,
"state": str,
"postal_code": str,
"customer_id": int,
"address_type": int,
"user_id": int,
"country_id": int,
"is_international": bool,
"province": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2Address":
"""Parse an lxml row element into this model with original_record set."""
kwargs: dict[str, Any] = {}
for hq_name, python_name in cls.get_columns():
child = row_elem.find(hq_name)
if child is not None and child.text is not None:
target_type = cls._COLUMN_TYPES.get(python_name, str)
if target_type is bool:
kwargs[python_name] = child.text.lower() in ("true", "1")
elif target_type in (int, float):
try:
kwargs[python_name] = target_type(child.text)
except (ValueError, TypeError):
kwargs[python_name] = child.text
else:
kwargs[python_name] = child.text
instance = cls(**kwargs)
instance.original_record = copy.deepcopy(instance)
return instance
[docs]
@dataclass
class TransactionRights:
"""
Represents a row in the TransactionRights table.
"""
transaction_type_id: int = 0
transaction_type: str = ""
description: str = ""
enabled: bool = False
authorize: bool = False
cancel: bool = False
create: bool = False
create_restricted: bool = False
update: bool = False
view: int = 0
limit_per_transaction: float = 0.0
limit_per_day: float = 0.0
limit_per_month: float = 0.0
limit_per_acct_per_day: float = 0.0
count_per_acct_per_day: int = 0
count_per_day: int = 0
count_per_month: int = 0
dual_approval_limit: float = 0.0
token_required_limit: float = 0.0
bit_flags: int = 0
original_record: Optional["TransactionRights"] = field(default=None, repr=False)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["TransactionTypeID", "transaction_type_id"],
["TransactionType", "transaction_type"],
["Description", "description"],
["Enabled", "enabled"],
["Authorize", "authorize"],
["Cancel", "cancel"],
["Create", "create"],
["CreateRestricted", "create_restricted"],
["Update", "update"],
["View", "view"],
["LimitPerTransaction", "limit_per_transaction"],
["LimitPerDay", "limit_per_day"],
["LimitPerMonth", "limit_per_month"],
["LimitPerAcctPerDay", "limit_per_acct_per_day"],
["CountPerAcctPerDay", "count_per_acct_per_day"],
["CountPerDay", "count_per_day"],
["CountPerMonth", "count_per_month"],
["DualApprovalLimit", "dual_approval_limit"],
["TokenRequiredLimit", "token_required_limit"],
["BitFlags", "bit_flags"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.transaction_type_id,
self.transaction_type,
self.description,
self.enabled,
self.authorize,
self.cancel,
self.create,
self.create_restricted,
self.update,
self.view,
self.limit_per_transaction,
self.limit_per_day,
self.limit_per_month,
self.limit_per_acct_per_day,
self.count_per_acct_per_day,
self.count_per_day,
self.count_per_month,
self.dual_approval_limit,
self.token_required_limit,
self.bit_flags,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"transaction_type_id": int,
"transaction_type": str,
"description": str,
"enabled": bool,
"authorize": bool,
"cancel": bool,
"create": bool,
"create_restricted": bool,
"update": bool,
"view": int,
"limit_per_transaction": float,
"limit_per_day": float,
"limit_per_month": float,
"limit_per_acct_per_day": float,
"count_per_acct_per_day": int,
"count_per_day": int,
"count_per_month": int,
"dual_approval_limit": float,
"token_required_limit": float,
"bit_flags": int,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "TransactionRights":
"""Parse an lxml row element into this model with original_record set."""
kwargs: dict[str, Any] = {}
for hq_name, python_name in cls.get_columns():
child = row_elem.find(hq_name)
if child is not None and child.text is not None:
target_type = cls._COLUMN_TYPES.get(python_name, str)
if target_type is bool:
kwargs[python_name] = child.text.lower() in ("true", "1")
elif target_type in (int, float):
try:
kwargs[python_name] = target_type(child.text)
except (ValueError, TypeError):
kwargs[python_name] = child.text
else:
kwargs[python_name] = child.text
instance = cls(**kwargs)
instance.original_record = copy.deepcopy(instance)
return instance
[docs]
@dataclass
class UserRights:
"""
Represents a row in the UserRights table.
"""
rights_short_name: str = ""
rights_long_name: str = ""
has_rights: bool = False
original_record: Optional["UserRights"] = field(default=None, repr=False)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["RightsShortName", "rights_short_name"],
["RightsLongName", "rights_long_name"],
["HasRights", "has_rights"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.rights_short_name,
self.rights_long_name,
self.has_rights,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"rights_short_name": str,
"rights_long_name": str,
"has_rights": bool,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "UserRights":
"""Parse an lxml row element into this model with original_record set."""
kwargs: dict[str, Any] = {}
for hq_name, python_name in cls.get_columns():
child = row_elem.find(hq_name)
if child is not None and child.text is not None:
target_type = cls._COLUMN_TYPES.get(python_name, str)
if target_type is bool:
kwargs[python_name] = child.text.lower() in ("true", "1")
elif target_type in (int, float):
try:
kwargs[python_name] = target_type(child.text)
except (ValueError, TypeError):
kwargs[python_name] = child.text
else:
kwargs[python_name] = child.text
instance = cls(**kwargs)
instance.original_record = copy.deepcopy(instance)
return instance
[docs]
@dataclass
class UserMonetaryRights:
"""
Represents a row in the UserMonetaryRights table.
"""
rights_short_name: str = ""
rights_long_name: str = ""
amount: float = 0.0
original_record: Optional["UserMonetaryRights"] = field(default=None, repr=False)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["RightsShortName", "rights_short_name"],
["RightsLongName", "rights_long_name"],
["Amount", "amount"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.rights_short_name,
self.rights_long_name,
self.amount,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"rights_short_name": str,
"rights_long_name": str,
"amount": float,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "UserMonetaryRights":
"""Parse an lxml row element into this model with original_record set."""
kwargs: dict[str, Any] = {}
for hq_name, python_name in cls.get_columns():
child = row_elem.find(hq_name)
if child is not None and child.text is not None:
target_type = cls._COLUMN_TYPES.get(python_name, str)
if target_type is bool:
kwargs[python_name] = child.text.lower() in ("true", "1")
elif target_type in (int, float):
try:
kwargs[python_name] = target_type(child.text)
except (ValueError, TypeError):
kwargs[python_name] = child.text
else:
kwargs[python_name] = child.text
instance = cls(**kwargs)
instance.original_record = copy.deepcopy(instance)
return instance
[docs]
@dataclass
class Q2CustomerAccountView:
"""
Represents a row in the Q2_CustomerAccountView table.
"""
customer_account_id: int = 0
customer_id: int = 0
is_cif_rel: Optional[bool] = None
host_account_id: int = 0
account_number_internal: str = ""
product_name: str = ""
is_user_enabled: Optional[bool] = None
is_external_account: bool = False
product_id: int = 0
product_type_id: int = 0
user_access: int = 0
current_access_flag: int = 0
valid_customer_access_flag: Optional[int] = None
valid_system_access_flag: Optional[int] = None
original_record: Optional["Q2CustomerAccountView"] = field(default=None, repr=False)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["CustomerAccountID", "customer_account_id"],
["CustomerID", "customer_id"],
["IsCifRel", "is_cif_rel"],
["HostAccountID", "host_account_id"],
["AccountNumberInternal", "account_number_internal"],
["ProductName", "product_name"],
["IsUserEnabled", "is_user_enabled"],
["IsExternalAccount", "is_external_account"],
["ProductID", "product_id"],
["ProductTypeID", "product_type_id"],
["UserAccess", "user_access"],
["CurrentAccessFlag", "current_access_flag"],
["ValidCustomerAccessFlag", "valid_customer_access_flag"],
["ValidSystemAccessFlag", "valid_system_access_flag"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.customer_account_id,
self.customer_id,
self.is_cif_rel,
self.host_account_id,
self.account_number_internal,
self.product_name,
self.is_user_enabled,
self.is_external_account,
self.product_id,
self.product_type_id,
self.user_access,
self.current_access_flag,
self.valid_customer_access_flag,
self.valid_system_access_flag,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"customer_account_id": int,
"customer_id": int,
"is_cif_rel": bool,
"host_account_id": int,
"account_number_internal": str,
"product_name": str,
"is_user_enabled": bool,
"is_external_account": bool,
"product_id": int,
"product_type_id": int,
"user_access": int,
"current_access_flag": int,
"valid_customer_access_flag": int,
"valid_system_access_flag": int,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2CustomerAccountView":
"""Parse an lxml row element into this model with original_record set."""
kwargs: dict[str, Any] = {}
for hq_name, python_name in cls.get_columns():
child = row_elem.find(hq_name)
if child is not None and child.text is not None:
target_type = cls._COLUMN_TYPES.get(python_name, str)
if target_type is bool:
kwargs[python_name] = child.text.lower() in ("true", "1")
elif target_type in (int, float):
try:
kwargs[python_name] = target_type(child.text)
except (ValueError, TypeError):
kwargs[python_name] = child.text
else:
kwargs[python_name] = child.text
instance = cls(**kwargs)
instance.original_record = copy.deepcopy(instance)
return instance
[docs]
def build_dal_customer_edit_dataset(
q2_customer: Optional[Q2Customer | list[Q2Customer]] = None,
q2_phone_number: Optional[Q2PhoneNumber | list[Q2PhoneNumber]] = None,
q2_customer_account: Optional[Q2CustomerAccount | list[Q2CustomerAccount]] = None,
q2_customer_account_access_view: Optional[
Q2CustomerAccountAccessView | list[Q2CustomerAccountAccessView]
] = None,
q2_subsidiary_view: Optional[Q2SubsidiaryView | list[Q2SubsidiaryView]] = None,
q2_address: Optional[Q2Address | list[Q2Address]] = None,
) -> DataSetBuilder:
"""
Build the dal_customer_edit DataSet parameter for AddCustomer.
For update operations, each model instance should have its original_record
set to the pre-modification state (from a prior GET call).
For add operations, set auto-increment ID fields to negative values.
:returns: DataSetBuilder ready to pass as the 'dal_customer_edit' parameter
"""
dataset = DataSetBuilder("DalCustomerEdit")
# Q2_Customer
q2_customer_cols = [col[0] for col in Q2Customer.get_columns()]
q2_customer_table = dataset.add_table("Q2_Customer", q2_customer_cols)
_add_rows_to_table(q2_customer_table, q2_customer)
# Q2_PhoneNumber
q2_phone_number_cols = [col[0] for col in Q2PhoneNumber.get_columns()]
q2_phone_number_table = dataset.add_table("Q2_PhoneNumber", q2_phone_number_cols)
_add_rows_to_table(q2_phone_number_table, q2_phone_number)
# Q2_CustomerAccount
q2_customer_account_cols = [col[0] for col in Q2CustomerAccount.get_columns()]
q2_customer_account_table = dataset.add_table(
"Q2_CustomerAccount", q2_customer_account_cols
)
_add_rows_to_table(q2_customer_account_table, q2_customer_account)
# Q2_CustomerAccountAccessView
q2_customer_account_access_view_cols = [
col[0] for col in Q2CustomerAccountAccessView.get_columns()
]
q2_customer_account_access_view_table = dataset.add_table(
"Q2_CustomerAccountAccessView", q2_customer_account_access_view_cols
)
_add_rows_to_table(
q2_customer_account_access_view_table, q2_customer_account_access_view
)
# Q2_SubsidiaryView
q2_subsidiary_view_cols = [col[0] for col in Q2SubsidiaryView.get_columns()]
q2_subsidiary_view_table = dataset.add_table(
"Q2_SubsidiaryView", q2_subsidiary_view_cols
)
_add_rows_to_table(q2_subsidiary_view_table, q2_subsidiary_view)
# Q2_Address
q2_address_cols = [col[0] for col in Q2Address.get_columns()]
q2_address_table = dataset.add_table("Q2_Address", q2_address_cols)
_add_rows_to_table(q2_address_table, q2_address)
return dataset
[docs]
def build_customer_rights_dataset(
transaction_rights: Optional[TransactionRights | list[TransactionRights]] = None,
user_rights: Optional[UserRights | list[UserRights]] = None,
user_monetary_rights: Optional[
UserMonetaryRights | list[UserMonetaryRights]
] = None,
q2_customer_account_view: Optional[
Q2CustomerAccountView | list[Q2CustomerAccountView]
] = None,
) -> DataSetBuilder:
"""
Build the customer_rights DataSet parameter for AddCustomer.
For update operations, each model instance should have its original_record
set to the pre-modification state (from a prior GET call).
For add operations, set auto-increment ID fields to negative values.
:returns: DataSetBuilder ready to pass as the 'customer_rights' parameter
"""
dataset = DataSetBuilder("UserRightsEdit")
# TransactionRights
transaction_rights_cols = [col[0] for col in TransactionRights.get_columns()]
transaction_rights_table = dataset.add_table(
"TransactionRights", transaction_rights_cols
)
_add_rows_to_table(transaction_rights_table, transaction_rights)
# UserRights
user_rights_cols = [col[0] for col in UserRights.get_columns()]
user_rights_table = dataset.add_table("UserRights", user_rights_cols)
_add_rows_to_table(user_rights_table, user_rights)
# UserMonetaryRights
user_monetary_rights_cols = [col[0] for col in UserMonetaryRights.get_columns()]
user_monetary_rights_table = dataset.add_table(
"UserMonetaryRights", user_monetary_rights_cols
)
_add_rows_to_table(user_monetary_rights_table, user_monetary_rights)
# Q2_CustomerAccountView
q2_customer_account_view_cols = [
col[0] for col in Q2CustomerAccountView.get_columns()
]
q2_customer_account_view_table = dataset.add_table(
"Q2_CustomerAccountView", q2_customer_account_view_cols
)
_add_rows_to_table(q2_customer_account_view_table, q2_customer_account_view)
return dataset
def _add_rows_to_table(
table: DataSetTable,
records: Optional[Any | list],
) -> None:
"""
Add rows to a DataSetTable from one or more model instances.
Handles Modified+Unchanged pairs (for updates), Added rows (for inserts),
and Deleted rows automatically based on each record's original_record field.
"""
if records is None:
return
if not isinstance(records, list):
records = [records]
for record in records:
values = record.to_row_values()
if record.original_record is not None:
# Update: emit Modified row (new values) + Unchanged row (original)
table.add_row(RowState.MODIFIED, values)
table.add_row(RowState.UNCHANGED, record.original_record.to_row_values())
else:
# Add: emit Added row (new record)
table.add_row(RowState.ADDED, values)