# This is an autogenerated file from the command "q2 generate_hq_api" and will be overwritten if run again
"""
DataSet models for AddRecipientsCse.
These typed dataclass models represent the tables and columns within the
DataSet parameters for the AddRecipientsCse 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 AddRecipientsCse
q2_recipient_account = AddRecipientsCse.Q2RecipientAccount(
recipient_id=1,
account_number="...",
account_type_id=1,
)
q2_recipient_address = AddRecipientsCse.Q2RecipientAddress(
recipient_id=1,
)
q2_recipient_fi_detail = AddRecipientsCse.Q2RecipientFIDetail(
account_id=1,
is_intermed_fi=True,
)
q2_recipient = AddRecipientsCse.Q2Recipient(
display_name="...",
customer_id=1,
always_send_email=True,
is_international=True,
ach_class_code_id=1,
)
q2_recipient_account_wage_garnishment = AddRecipientsCse.Q2RecipientAccountWageGarnishment(
account_id=1,
disbursement_unit_name="...",
tax_type_short_name="...",
entry_type_short_name="...",
)
q2_recipient_pending = AddRecipientsCse.Q2RecipientPending(
recipient_id=1,
change_status="...",
customer_id_=1,
created_at="...",
)
q2_user = AddRecipientsCse.Q2User(
customer_id=1,
create_date="...",
auto_generated=True,
use_customer_accounts=True,
user_status_id=1,
)
q2_recipient_fi_additional_detail = AddRecipientsCse.Q2RecipientFiAdditionalDetail(
recipient_fi_detail_id=1,
is_iat_enabled=True,
is_wire_enabled=True,
)
"""
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 Q2RecipientAccount:
"""
Represents a row in the Q2_RecipientAccount table.
"""
account_id: int = -1
recipient_id: int = 0
account_number: str = ""
account_type_id: int = 0
aba: Optional[str] = None
wage_garnishment_tax_type_code_id: Optional[int] = None
wage_garnishment_addenda: Optional[str] = None
rtp_aba: Optional[str] = None
allowed_rtp_payment_rails: Optional[int] = None
pre_note_expiration: Optional[str] = None
pending_pre_note_expiration: Optional[str] = None
wire_fi_id: Optional[str] = None
wire_type_id: Optional[int] = None
original_record: Optional["Q2RecipientAccount"] = field(default=None, repr=False)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["AccountID", "account_id"],
["RecipientID", "recipient_id"],
["AccountNumber", "account_number"],
["AccountTypeID", "account_type_id"],
["ABA", "aba"],
["WageGarnishmentTaxTypeCodeID", "wage_garnishment_tax_type_code_id"],
["WageGarnishmentAddenda", "wage_garnishment_addenda"],
["RTPAba", "rtp_aba"],
["AllowedRtpPaymentRails", "allowed_rtp_payment_rails"],
["PreNoteExpiration", "pre_note_expiration"],
["PendingPreNoteExpiration", "pending_pre_note_expiration"],
["WireFiID", "wire_fi_id"],
["WireTypeID", "wire_type_id"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.account_id,
self.recipient_id,
self.account_number,
self.account_type_id,
self.aba,
self.wage_garnishment_tax_type_code_id,
self.wage_garnishment_addenda,
self.rtp_aba,
self.allowed_rtp_payment_rails,
self.pre_note_expiration,
self.pending_pre_note_expiration,
self.wire_fi_id,
self.wire_type_id,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"account_id": int,
"recipient_id": int,
"account_number": str,
"account_type_id": int,
"aba": str,
"wage_garnishment_tax_type_code_id": int,
"wage_garnishment_addenda": str,
"rtp_aba": str,
"allowed_rtp_payment_rails": int,
"pre_note_expiration": str,
"pending_pre_note_expiration": str,
"wire_fi_id": str,
"wire_type_id": int,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2RecipientAccount":
"""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 Q2RecipientAddress:
"""
Represents a row in the Q2_RecipientAddress table.
"""
recipient_address_id: int = -1
recipient_id: int = 0
address1: Optional[str] = None
address2: Optional[str] = None
city: Optional[str] = None
state: Optional[str] = None
postal_code: Optional[str] = None
country_id: Optional[int] = None
address3: Optional[str] = None
province: Optional[str] = None
original_record: Optional["Q2RecipientAddress"] = field(default=None, repr=False)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["RecipientAddressID", "recipient_address_id"],
["RecipientID", "recipient_id"],
["Address1", "address1"],
["Address2", "address2"],
["City", "city"],
["State", "state"],
["PostalCode", "postal_code"],
["CountryID", "country_id"],
["Address3", "address3"],
["Province", "province"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.recipient_address_id,
self.recipient_id,
self.address1,
self.address2,
self.city,
self.state,
self.postal_code,
self.country_id,
self.address3,
self.province,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"recipient_address_id": int,
"recipient_id": int,
"address1": str,
"address2": str,
"city": str,
"state": str,
"postal_code": str,
"country_id": int,
"address3": str,
"province": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2RecipientAddress":
"""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 Q2RecipientFIDetail:
"""
Represents a row in the Q2_RecipientFIDetail table.
"""
recipient_fi_detail_id: int = -1
account_id: int = 0
name: Optional[str] = None
bic: Optional[str] = None
branch_bic: Optional[str] = None
address1: Optional[str] = None
address2: Optional[str] = None
address3: Optional[str] = None
city: Optional[str] = None
state: Optional[str] = None
postal_code: Optional[str] = None
country_id: Optional[int] = None
intermediary_aba: Optional[str] = None
is_intermed_fi: bool = False
iban: Optional[str] = None
receiving_fi_short_name: Optional[str] = None
receiving_fi_aba: Optional[str] = None
account_num_between_banks: Optional[str] = None
original_record: Optional["Q2RecipientFIDetail"] = field(default=None, repr=False)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["RecipientFIDetailID", "recipient_fi_detail_id"],
["AccountID", "account_id"],
["Name", "name"],
["BIC", "bic"],
["BranchBIC", "branch_bic"],
["Address1", "address1"],
["Address2", "address2"],
["Address3", "address3"],
["City", "city"],
["State", "state"],
["PostalCode", "postal_code"],
["CountryID", "country_id"],
["IntermediaryABA", "intermediary_aba"],
["IsIntermedFI", "is_intermed_fi"],
["IBAN", "iban"],
["ReceivingFiShortName", "receiving_fi_short_name"],
["ReceivingFiAba", "receiving_fi_aba"],
["AccountNumBetweenBanks", "account_num_between_banks"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.recipient_fi_detail_id,
self.account_id,
self.name,
self.bic,
self.branch_bic,
self.address1,
self.address2,
self.address3,
self.city,
self.state,
self.postal_code,
self.country_id,
self.intermediary_aba,
self.is_intermed_fi,
self.iban,
self.receiving_fi_short_name,
self.receiving_fi_aba,
self.account_num_between_banks,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"recipient_fi_detail_id": int,
"account_id": int,
"name": str,
"bic": str,
"branch_bic": str,
"address1": str,
"address2": str,
"address3": str,
"city": str,
"state": str,
"postal_code": str,
"country_id": int,
"intermediary_aba": str,
"is_intermed_fi": bool,
"iban": str,
"receiving_fi_short_name": str,
"receiving_fi_aba": str,
"account_num_between_banks": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2RecipientFIDetail":
"""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 Q2Recipient:
"""
Represents a row in the Q2_Recipient table.
"""
recipient_id: int = -1
display_name: str = ""
customer_id: int = 0
default_account_id: Optional[int] = None
email_address: Optional[str] = None
always_send_email: bool = False
is_international: bool = False
ach_class_code_id: int = 0
ach_name: Optional[str] = None
wire_name: Optional[str] = None
identification_number: Optional[str] = None
is_individual: Optional[bool] = None
rtp_name: Optional[str] = None
legal_entity_identifier: Optional[str] = None
original_record: Optional["Q2Recipient"] = field(default=None, repr=False)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["RecipientID", "recipient_id"],
["DisplayName", "display_name"],
["CustomerID", "customer_id"],
["DefaultAccountID", "default_account_id"],
["EmailAddress", "email_address"],
["AlwaysSendEmail", "always_send_email"],
["IsInternational", "is_international"],
["ACHClassCodeID", "ach_class_code_id"],
["AchName", "ach_name"],
["WireName", "wire_name"],
["IdentificationNumber", "identification_number"],
["IsIndividual", "is_individual"],
["RTPName", "rtp_name"],
["LegalEntityIdentifier", "legal_entity_identifier"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.recipient_id,
self.display_name,
self.customer_id,
self.default_account_id,
self.email_address,
self.always_send_email,
self.is_international,
self.ach_class_code_id,
self.ach_name,
self.wire_name,
self.identification_number,
self.is_individual,
self.rtp_name,
self.legal_entity_identifier,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"recipient_id": int,
"display_name": str,
"customer_id": int,
"default_account_id": int,
"email_address": str,
"always_send_email": bool,
"is_international": bool,
"ach_class_code_id": int,
"ach_name": str,
"wire_name": str,
"identification_number": str,
"is_individual": bool,
"rtp_name": str,
"legal_entity_identifier": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2Recipient":
"""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 Q2RecipientAccountWageGarnishment:
"""
Represents a row in the Q2_RecipientAccountWageGarnishment table.
"""
wage_garnishment_id: int = -1
account_id: int = 0
disbursement_unit_name: str = ""
tax_type_short_name: str = ""
entry_type_short_name: str = ""
case_identifier: Optional[str] = None
garnished_ssn: Optional[str] = None
medical_support_indicator: Optional[bool] = None
originator_reference_id: Optional[str] = None
additional_elements: Optional[str] = None
original_record: Optional["Q2RecipientAccountWageGarnishment"] = field(
default=None, repr=False
)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["WageGarnishmentID", "wage_garnishment_id"],
["AccountID", "account_id"],
["DisbursementUnitName", "disbursement_unit_name"],
["TaxTypeShortName", "tax_type_short_name"],
["EntryTypeShortName", "entry_type_short_name"],
["CaseIdentifier", "case_identifier"],
["GarnishedSSN", "garnished_ssn"],
["MedicalSupportIndicator", "medical_support_indicator"],
["OriginatorReferenceId", "originator_reference_id"],
["AdditionalElements", "additional_elements"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.wage_garnishment_id,
self.account_id,
self.disbursement_unit_name,
self.tax_type_short_name,
self.entry_type_short_name,
self.case_identifier,
self.garnished_ssn,
self.medical_support_indicator,
self.originator_reference_id,
self.additional_elements,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"wage_garnishment_id": int,
"account_id": int,
"disbursement_unit_name": str,
"tax_type_short_name": str,
"entry_type_short_name": str,
"case_identifier": str,
"garnished_ssn": str,
"medical_support_indicator": bool,
"originator_reference_id": str,
"additional_elements": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2RecipientAccountWageGarnishment":
"""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 Q2RecipientPending:
"""
Represents a row in the Q2_RecipientPending table.
"""
pending_recipient_id: int = -1
recipient_id: int = 0
change_status: str = ""
data: Optional[str] = None
customer_id_: int = 0
user_id: Optional[int] = None
created_at: str = ""
approved: Optional[bool] = None
approved_id_: Optional[int] = None
approved_at: Optional[str] = None
approval_reason: Optional[str] = None
noc_id: Optional[str] = None
original_record: Optional["Q2RecipientPending"] = field(default=None, repr=False)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["PendingRecipientID", "pending_recipient_id"],
["RecipientID", "recipient_id"],
["ChangeStatus", "change_status"],
["Data", "data"],
["CustomerID_", "customer_id_"],
["UserID", "user_id"],
["CreatedAt", "created_at"],
["Approved", "approved"],
["ApprovedID_", "approved_id_"],
["ApprovedAt", "approved_at"],
["ApprovalReason", "approval_reason"],
["NocID", "noc_id"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.pending_recipient_id,
self.recipient_id,
self.change_status,
self.data,
self.customer_id_,
self.user_id,
self.created_at,
self.approved,
self.approved_id_,
self.approved_at,
self.approval_reason,
self.noc_id,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"pending_recipient_id": int,
"recipient_id": int,
"change_status": str,
"data": str,
"customer_id_": int,
"user_id": int,
"created_at": str,
"approved": bool,
"approved_id_": int,
"approved_at": str,
"approval_reason": str,
"noc_id": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2RecipientPending":
"""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 Q2User:
"""
Represents a row in the Q2_User table.
"""
user_id: int = -1
customer_id: int = 0
first_name: Optional[str] = None
middle_name: Optional[str] = None
last_name: Optional[str] = None
salutation: Optional[str] = None
suffix: Optional[str] = None
ssn: Optional[str] = None
create_date: str = ""
deleted_date: Optional[str] = None
default_email_id: Optional[int] = None
default_address_id: Optional[int] = None
auto_generated: bool = False
host_user: Optional[str] = None
host_pwd: Optional[str] = None
user_info: Optional[str] = None
profile_updated: Optional[bool] = None
mobile_auth_code: Optional[str] = None
use_customer_accounts: bool = False
primary_cif: Optional[str] = None
last_user_profile_update: Optional[str] = None
user_role_id: Optional[int] = None
policy_id: Optional[int] = None
user_status_id: int = 0
status_change_date: Optional[str] = None
dob: Optional[str] = None
original_record: Optional["Q2User"] = field(default=None, repr=False)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["UserID", "user_id"],
["CustomerID", "customer_id"],
["FirstName", "first_name"],
["MiddleName", "middle_name"],
["LastName", "last_name"],
["Salutation", "salutation"],
["Suffix", "suffix"],
["SSN", "ssn"],
["CreateDate", "create_date"],
["DeletedDate", "deleted_date"],
["DefaultEmailID", "default_email_id"],
["DefaultAddressID", "default_address_id"],
["AutoGenerated", "auto_generated"],
["HostUser", "host_user"],
["HostPwd", "host_pwd"],
["UserInfo", "user_info"],
["ProfileUpdated", "profile_updated"],
["MobileAuthCode", "mobile_auth_code"],
["UseCustomerAccounts", "use_customer_accounts"],
["PrimaryCIF", "primary_cif"],
["LastUserProfileUpdate", "last_user_profile_update"],
["UserRoleID", "user_role_id"],
["PolicyID", "policy_id"],
["UserStatusID", "user_status_id"],
["StatusChangeDate", "status_change_date"],
["DOB", "dob"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.user_id,
self.customer_id,
self.first_name,
self.middle_name,
self.last_name,
self.salutation,
self.suffix,
self.ssn,
self.create_date,
self.deleted_date,
self.default_email_id,
self.default_address_id,
self.auto_generated,
self.host_user,
self.host_pwd,
self.user_info,
self.profile_updated,
self.mobile_auth_code,
self.use_customer_accounts,
self.primary_cif,
self.last_user_profile_update,
self.user_role_id,
self.policy_id,
self.user_status_id,
self.status_change_date,
self.dob,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"user_id": int,
"customer_id": int,
"first_name": str,
"middle_name": str,
"last_name": str,
"salutation": str,
"suffix": str,
"ssn": str,
"create_date": str,
"deleted_date": str,
"default_email_id": int,
"default_address_id": int,
"auto_generated": bool,
"host_user": str,
"host_pwd": str,
"user_info": str,
"profile_updated": bool,
"mobile_auth_code": str,
"use_customer_accounts": bool,
"primary_cif": str,
"last_user_profile_update": str,
"user_role_id": int,
"policy_id": int,
"user_status_id": int,
"status_change_date": str,
"dob": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2User":
"""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 Q2RecipientFiAdditionalDetail:
"""
Represents a row in the Q2_RecipientFiAdditionalDetail table.
"""
recipient_fi_additional_detail_id: int = -1
recipient_fi_detail_id: int = 0
transit_number: Optional[str] = None
is_iat_enabled: bool = False
is_wire_enabled: bool = False
original_record: Optional["Q2RecipientFiAdditionalDetail"] = field(
default=None, repr=False
)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["RecipientFiAdditionalDetailID", "recipient_fi_additional_detail_id"],
["RecipientFIDetailID", "recipient_fi_detail_id"],
["TransitNumber", "transit_number"],
["IsIatEnabled", "is_iat_enabled"],
["IsWireEnabled", "is_wire_enabled"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.recipient_fi_additional_detail_id,
self.recipient_fi_detail_id,
self.transit_number,
self.is_iat_enabled,
self.is_wire_enabled,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"recipient_fi_additional_detail_id": int,
"recipient_fi_detail_id": int,
"transit_number": str,
"is_iat_enabled": bool,
"is_wire_enabled": bool,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2RecipientFiAdditionalDetail":
"""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_recipients_dataset(
q2_recipient_account: Optional[
Q2RecipientAccount | list[Q2RecipientAccount]
] = None,
q2_recipient_address: Optional[
Q2RecipientAddress | list[Q2RecipientAddress]
] = None,
q2_recipient_fi_detail: Optional[
Q2RecipientFIDetail | list[Q2RecipientFIDetail]
] = None,
q2_recipient: Optional[Q2Recipient | list[Q2Recipient]] = None,
q2_recipient_account_wage_garnishment: Optional[
Q2RecipientAccountWageGarnishment | list[Q2RecipientAccountWageGarnishment]
] = None,
q2_recipient_pending: Optional[
Q2RecipientPending | list[Q2RecipientPending]
] = None,
q2_user: Optional[Q2User | list[Q2User]] = None,
q2_recipient_fi_additional_detail: Optional[
Q2RecipientFiAdditionalDetail | list[Q2RecipientFiAdditionalDetail]
] = None,
) -> DataSetBuilder:
"""
Build the recipients DataSet parameter for AddRecipientsCse.
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 'recipients' parameter
"""
dataset = DataSetBuilder("DalRecipientEdit")
# Q2_RecipientAccount
q2_recipient_account_cols = [col[0] for col in Q2RecipientAccount.get_columns()]
q2_recipient_account_table = dataset.add_table(
"Q2_RecipientAccount", q2_recipient_account_cols
)
_add_rows_to_table(q2_recipient_account_table, q2_recipient_account)
# Q2_RecipientAddress
q2_recipient_address_cols = [col[0] for col in Q2RecipientAddress.get_columns()]
q2_recipient_address_table = dataset.add_table(
"Q2_RecipientAddress", q2_recipient_address_cols
)
_add_rows_to_table(q2_recipient_address_table, q2_recipient_address)
# Q2_RecipientFIDetail
q2_recipient_fi_detail_cols = [col[0] for col in Q2RecipientFIDetail.get_columns()]
q2_recipient_fi_detail_table = dataset.add_table(
"Q2_RecipientFIDetail", q2_recipient_fi_detail_cols
)
_add_rows_to_table(q2_recipient_fi_detail_table, q2_recipient_fi_detail)
# Q2_Recipient
q2_recipient_cols = [col[0] for col in Q2Recipient.get_columns()]
q2_recipient_table = dataset.add_table("Q2_Recipient", q2_recipient_cols)
_add_rows_to_table(q2_recipient_table, q2_recipient)
# Q2_RecipientAccountWageGarnishment
q2_recipient_account_wage_garnishment_cols = [
col[0] for col in Q2RecipientAccountWageGarnishment.get_columns()
]
q2_recipient_account_wage_garnishment_table = dataset.add_table(
"Q2_RecipientAccountWageGarnishment", q2_recipient_account_wage_garnishment_cols
)
_add_rows_to_table(
q2_recipient_account_wage_garnishment_table,
q2_recipient_account_wage_garnishment,
)
# Q2_RecipientPending
q2_recipient_pending_cols = [col[0] for col in Q2RecipientPending.get_columns()]
q2_recipient_pending_table = dataset.add_table(
"Q2_RecipientPending", q2_recipient_pending_cols
)
_add_rows_to_table(q2_recipient_pending_table, q2_recipient_pending)
# Q2_User
q2_user_cols = [col[0] for col in Q2User.get_columns()]
q2_user_table = dataset.add_table("Q2_User", q2_user_cols)
_add_rows_to_table(q2_user_table, q2_user)
# Q2_RecipientFiAdditionalDetail
q2_recipient_fi_additional_detail_cols = [
col[0] for col in Q2RecipientFiAdditionalDetail.get_columns()
]
q2_recipient_fi_additional_detail_table = dataset.add_table(
"Q2_RecipientFiAdditionalDetail", q2_recipient_fi_additional_detail_cols
)
_add_rows_to_table(
q2_recipient_fi_additional_detail_table, q2_recipient_fi_additional_detail
)
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)