# This is an autogenerated file from the command "q2 generate_hq_api" and will be overwritten if run again
"""
DataSet models for ProcessGeneratedTransactions.
These typed dataclass models represent the tables and columns within the
DataSet parameters for the ProcessGeneratedTransactions 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 ProcessGeneratedTransactions
q2_generated_transaction_view = ProcessGeneratedTransactions.Q2GeneratedTransactionView(
transaction_id=1,
customer_id=1,
customer_name="...",
user_id=1,
create_date="...",
ui_source_id=1,
ui_source="...",
transaction_status_id=1,
transaction_status="...",
transaction_status_description="...",
transaction_type_id=1,
transaction_type="...",
transaction_type_description="...",
diagnostic=True,
group_id=1,
group_desc="...",
oob_auth_required=True,
oob_auth_provided=True,
is_reversed=True,
)
q2_change_of_address_transaction_accounts = ProcessGeneratedTransactions.Q2ChangeOfAddressTransactionAccounts(
transaction_id=1,
host_account_id=1,
account_number_internal="...",
product_id=1,
product_name="...",
hydra_product_code="...",
host_product_code="...",
product_type_id=1,
hydra_product_type_code="...",
)
q2_change_of_address_transactions = ProcessGeneratedTransactions.Q2ChangeOfAddressTransactions(
transaction_id=1,
)
q2_check_re_order_transactions = ProcessGeneratedTransactions.Q2CheckReOrderTransactions(
)
q2_funds_transfer_transactions = ProcessGeneratedTransactions.Q2FundsTransferTransactions(
transaction_id=1,
account_number_internal="...",
product_id=1,
product_name="...",
hydra_product_code="...",
host_product_code="...",
product_type_id=1,
product_type_name="...",
hydra_product_type_code="...",
is_external_account=True,
)
q2_stop_payment_transactions = ProcessGeneratedTransactions.Q2StopPaymentTransactions(
transaction_id=1,
)
q2_bill_payment_transactions = ProcessGeneratedTransactions.Q2BillPaymentTransactions(
transaction_id=1,
payee_name="...",
account_number_or_memo="...",
pay_as_name="...",
pay_as_address1="...",
pay_as_city="...",
pay_as_state="...",
pay_as_postal_code="...",
is_international=True,
name="...",
payee_type="...",
)
q2_ach_ppd_ccd_transactions = ProcessGeneratedTransactions.Q2AchPpdCcdTransactions(
transaction_id=1,
company_name="...",
company_identification="...",
ach_class_code_id=1,
ach_class_code="...",
company_entry_description="...",
effective_entry_date="...",
transaction_type_id=1,
is_same_day_ach=True,
)
q2_ach_iat_transactions = ProcessGeneratedTransactions.Q2AchIatTransactions(
ach_iat_batch_id=1,
transaction_id=1,
transaction_type_id=1,
originator_identification="...",
company_entry_description="...",
effective_entry_date="...",
recipient_count=1,
service_class_code="...",
foreign_exchange_indicator="...",
destination_currency_code="...",
)
q2_wire_domestic_transactions = ProcessGeneratedTransactions.Q2WireDomesticTransactions(
transaction_id=1,
wire_transfer_id=1,
to_account="...",
to_account_type="...",
recip_country_name="...",
rec_fi_country_name="...",
)
q2_external_transfer_transactions = ProcessGeneratedTransactions.Q2ExternalTransferTransactions(
transaction_id=1,
is_same_day_ach=True,
account_number_internal="...",
product_id=1,
product_name="...",
hydra_product_code="...",
to_host_product_code="...",
product_type_id=1,
product_type_name="...",
hydra_product_type_code="...",
is_from_account_external=True,
from_host_product_code="...",
is_to_account_external=True,
)
q2_wire_international_transactions = ProcessGeneratedTransactions.Q2WireInternationalTransactions(
transaction_id=1,
wire_transfer_id=1,
to_account="...",
to_account_type="...",
rate_applied_to_usd_equivalent=True,
)
q2_send_check_transactions = ProcessGeneratedTransactions.Q2SendCheckTransactions(
transaction_id=1,
send_to_acct_owner=True,
is_cashier_or_certified_check=True,
)
q2_ach_pass_thru_transactions = ProcessGeneratedTransactions.Q2AchPassThruTransactions(
ach_pass_thru_id=1,
transaction_id=1,
transaction_type_id=1,
ach_file="...",
)
q2_rtp_get_transactions = ProcessGeneratedTransactions.Q2RTPGetTransactions(
child_bitflag=1,
create_date="...",
currency_code="...",
customer_id=1,
diagnostic=True,
host_product_code="...",
is_reversed=True,
oob_auth_provided=True,
oob_auth_required=True,
transaction_status_id=1,
transaction_type_id=1,
ui_source_id=1,
user_id=1,
outgoing=True,
local_entity_name="...",
local_entity_tax_id="...",
remote_entity_name="...",
remote_fi_account_number="...",
remote_fi_routing_number="...",
rtp_request_type_short_name="...",
created_at="...",
rtp_vendor_short_name="...",
)
"""
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 Q2GeneratedTransactionView:
"""
Represents a row in the Q2_GeneratedTransactionView table.
"""
transaction_id: int = 0
customer_id: int = 0
customer_name: str = ""
user_id: int = 0
user_name: Optional[str] = None
create_date: str = ""
ui_source_id: int = 0
ui_source: str = ""
transaction_status_id: int = 0
transaction_status: str = ""
transaction_status_description: str = ""
transaction_type_id: int = 0
transaction_type: str = ""
transaction_type_description: str = ""
process_date: Optional[str] = None
processed_date: Optional[str] = None
diagnostic: bool = False
prefunding_status: Optional[str] = None
prefunding_reason: Optional[str] = None
host_account_id: Optional[int] = None
account_number_internal: Optional[str] = None
account_number_external: Optional[str] = None
product_id: Optional[int] = None
product_name: Optional[str] = None
product_type_id: Optional[int] = None
product_type_name: Optional[str] = None
transaction_amount: Optional[float] = None
currency_code: Optional[str] = None
authorized_by_id: Optional[int] = None
authorized_by: Optional[str] = None
authorized_date: Optional[str] = None
modified_by_id: Optional[int] = None
modified_by: Optional[str] = None
modified_date: Optional[str] = None
recurring_transaction_id: Optional[int] = None
description: Optional[str] = None
template_id: Optional[int] = None
host_result: Optional[str] = None
hydra_product_type_code: Optional[str] = None
host_product_type_code: Optional[str] = None
hydra_product_code: Optional[str] = None
host_product_code: Optional[str] = None
processing_type_id: Optional[int] = None
processing_type: Optional[str] = None
processing_type_description: Optional[str] = None
target_description: Optional[str] = None
parent_id: Optional[int] = None
settlement_id: Optional[int] = None
has_children: Optional[bool] = None
is_distributed_funding_child: Optional[bool] = None
is_settlement_funding_child: Optional[bool] = None
skip_balance_check: Optional[bool] = None
cif_internal: Optional[str] = None
account_nick_name: Optional[str] = None
effective_date: Optional[str] = None
bill_payee_id: Optional[int] = None
payee_display_name: Optional[str] = None
payee_name: Optional[str] = None
audit_score_id: Optional[int] = None
tran_score_status: Optional[int] = None
tran_score_weight: Optional[int] = None
login_score_id: Optional[int] = None
login_score_status: Optional[int] = None
login_score_weight: Optional[int] = None
group_id: int = 0
group_desc: str = ""
to_host_account_id: Optional[int] = None
to_account_number_internal: Optional[str] = None
to_account_number_external: Optional[str] = None
to_product_name: Optional[str] = None
to_product_type_name: Optional[str] = None
has_case_notes: Optional[bool] = None
score_audit_id: Optional[int] = None
authorizing_user_credentials: Optional[str] = None
gam_rights_bit_flag: Optional[int] = None
user_role_id: Optional[int] = None
recipient_count: Optional[int] = None
subsidiary_id: Optional[int] = None
gt_batch_id: Optional[int] = None
recipient_name: Optional[str] = None
intermediary_bank_name: Optional[str] = None
beneficiary_bank_name: Optional[str] = None
message_to_beneficiary: Optional[str] = None
template_name: Optional[str] = None
template_description: Optional[str] = None
oob_auth_required: bool = False
oob_auth_provided: bool = False
recipient_account_number: Optional[str] = None
recipient_account_type: Optional[str] = None
principal: Optional[float] = None
interest: Optional[float] = None
usd_equivalent_amount: Optional[float] = None
recipient_identification_number: Optional[str] = None
is_same_day_ach: Optional[bool] = None
company_entry_description: Optional[str] = None
purpose_of_wire: Optional[str] = None
ach_class_code: Optional[str] = None
has_wage_garnishment_payments: Optional[bool] = None
replaces_transaction_id: Optional[int] = None
replaced_by_transaction_id: Optional[int] = None
is_reversed: bool = False
reversed_gtid: Optional[int] = None
is_reversible: Optional[bool] = None
to_product_id: Optional[int] = None
to_hydra_product_type_code: Optional[str] = None
to_hydra_product_code: Optional[str] = None
to_host_product_type_code: Optional[str] = None
to_host_product_code: Optional[str] = None
is_production: Optional[bool] = None
zone_id: Optional[int] = None
zone: Optional[str] = None
is_ach_iat: Optional[bool] = None
file_name: Optional[str] = None
gt_flavor_id: Optional[int] = None
gt_flavor_short_name: Optional[str] = None
recurring_master: Optional[bool] = None
real_time_payment_id: Optional[int] = None
rtp_thread_id: Optional[int] = None
rtp_outgoing: Optional[bool] = None
rtp_request_type_short_name: Optional[str] = None
rtp_vendor_short_name: Optional[str] = None
expiry_date: Optional[str] = None
vendor_status: Optional[str] = None
vendor_status_date_time: Optional[str] = None
service_call_unique_identifier: Optional[str] = None
service_call_conversation_identifier: Optional[str] = None
subsidiary_name: Optional[str] = None
is_large_batch_workflow: Optional[bool] = None
is_aggregate: Optional[bool] = None
cif_external: Optional[str] = None
host_branch: Optional[str] = None
tax_version_id: Optional[str] = None
file_delivery_status_id: Optional[int] = None
file_delivery_status: Optional[str] = None
file_delivery_date_time: Optional[str] = None
original_record: Optional["Q2GeneratedTransactionView"] = field(
default=None, repr=False
)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["TransactionID", "transaction_id"],
["CustomerID", "customer_id"],
["CustomerName", "customer_name"],
["UserID", "user_id"],
["UserName", "user_name"],
["CreateDate", "create_date"],
["UISourceID", "ui_source_id"],
["UISource", "ui_source"],
["TransactionStatusID", "transaction_status_id"],
["TransactionStatus", "transaction_status"],
["TransactionStatusDescription", "transaction_status_description"],
["TransactionTypeID", "transaction_type_id"],
["TransactionType", "transaction_type"],
["TransactionTypeDescription", "transaction_type_description"],
["ProcessDate", "process_date"],
["ProcessedDate", "processed_date"],
["Diagnostic", "diagnostic"],
["PrefundingStatus", "prefunding_status"],
["PrefundingReason", "prefunding_reason"],
["HostAccountID", "host_account_id"],
["AccountNumberInternal", "account_number_internal"],
["AccountNumberExternal", "account_number_external"],
["ProductID", "product_id"],
["ProductName", "product_name"],
["ProductTypeID", "product_type_id"],
["ProductTypeName", "product_type_name"],
["TransactionAmount", "transaction_amount"],
["CurrencyCode", "currency_code"],
["AuthorizedByID", "authorized_by_id"],
["AuthorizedBy", "authorized_by"],
["AuthorizedDate", "authorized_date"],
["ModifiedByID", "modified_by_id"],
["ModifiedBy", "modified_by"],
["ModifiedDate", "modified_date"],
["RecurringTransactionID", "recurring_transaction_id"],
["Description", "description"],
["TemplateID", "template_id"],
["HostResult", "host_result"],
["HydraProductTypeCode", "hydra_product_type_code"],
["HostProductTypeCode", "host_product_type_code"],
["HydraProductCode", "hydra_product_code"],
["HostProductCode", "host_product_code"],
["ProcessingTypeID", "processing_type_id"],
["ProcessingType", "processing_type"],
["ProcessingTypeDescription", "processing_type_description"],
["TargetDescription", "target_description"],
["ParentID", "parent_id"],
["SettlementID", "settlement_id"],
["HasChildren", "has_children"],
["IsDistributedFundingChild", "is_distributed_funding_child"],
["IsSettlementFundingChild", "is_settlement_funding_child"],
["SkipBalanceCheck", "skip_balance_check"],
["CIFInternal", "cif_internal"],
["AccountNickName", "account_nick_name"],
["EffectiveDate", "effective_date"],
["BillPayeeID", "bill_payee_id"],
["PayeeDisplayName", "payee_display_name"],
["PayeeName", "payee_name"],
["AuditScoreID", "audit_score_id"],
["TranScoreStatus", "tran_score_status"],
["TranScoreWeight", "tran_score_weight"],
["LoginScoreID", "login_score_id"],
["LoginScoreStatus", "login_score_status"],
["LoginScoreWeight", "login_score_weight"],
["GroupID", "group_id"],
["GroupDesc", "group_desc"],
["ToHostAccountID", "to_host_account_id"],
["ToAccountNumberInternal", "to_account_number_internal"],
["ToAccountNumberExternal", "to_account_number_external"],
["ToProductName", "to_product_name"],
["ToProductTypeName", "to_product_type_name"],
["HasCaseNotes", "has_case_notes"],
["ScoreAuditID", "score_audit_id"],
["AuthorizingUserCredentials", "authorizing_user_credentials"],
["GamRightsBitFlag", "gam_rights_bit_flag"],
["UserRoleID", "user_role_id"],
["RecipientCount", "recipient_count"],
["SubsidiaryID", "subsidiary_id"],
["GTBatchID", "gt_batch_id"],
["RecipientName", "recipient_name"],
["IntermediaryBankName", "intermediary_bank_name"],
["BeneficiaryBankName", "beneficiary_bank_name"],
["MessageToBeneficiary", "message_to_beneficiary"],
["TemplateName", "template_name"],
["TemplateDescription", "template_description"],
["OobAuthRequired", "oob_auth_required"],
["OobAuthProvided", "oob_auth_provided"],
["RecipientAccountNumber", "recipient_account_number"],
["RecipientAccountType", "recipient_account_type"],
["Principal", "principal"],
["Interest", "interest"],
["UsdEquivalentAmount", "usd_equivalent_amount"],
["RecipientIdentificationNumber", "recipient_identification_number"],
["IsSameDayAch", "is_same_day_ach"],
["CompanyEntryDescription", "company_entry_description"],
["PurposeOfWire", "purpose_of_wire"],
["AchClassCode", "ach_class_code"],
["HasWageGarnishmentPayments", "has_wage_garnishment_payments"],
["ReplacesTransactionID", "replaces_transaction_id"],
["ReplacedByTransactionID", "replaced_by_transaction_id"],
["IsReversed", "is_reversed"],
["ReversedGTID", "reversed_gtid"],
["IsReversible", "is_reversible"],
["ToProductID", "to_product_id"],
["ToHydraProductTypeCode", "to_hydra_product_type_code"],
["ToHydraProductCode", "to_hydra_product_code"],
["ToHostProductTypeCode", "to_host_product_type_code"],
["ToHostProductCode", "to_host_product_code"],
["IsProduction", "is_production"],
["ZoneID", "zone_id"],
["Zone", "zone"],
["IsAchIat", "is_ach_iat"],
["FileName", "file_name"],
["GtFlavorID", "gt_flavor_id"],
["GtFlavorShortName", "gt_flavor_short_name"],
["RecurringMaster", "recurring_master"],
["RealTimePaymentID", "real_time_payment_id"],
["RTPThreadID", "rtp_thread_id"],
["RTPOutgoing", "rtp_outgoing"],
["RTPRequestTypeShortName", "rtp_request_type_short_name"],
["RTPVendorShortName", "rtp_vendor_short_name"],
["ExpiryDate", "expiry_date"],
["VendorStatus", "vendor_status"],
["VendorStatusDateTime", "vendor_status_date_time"],
["ServiceCallUniqueIdentifier", "service_call_unique_identifier"],
[
"ServiceCallConversationIdentifier",
"service_call_conversation_identifier",
],
["SubsidiaryName", "subsidiary_name"],
["IsLargeBatchWorkflow", "is_large_batch_workflow"],
["IsAggregate", "is_aggregate"],
["CIFExternal", "cif_external"],
["HostBranch", "host_branch"],
["TaxVersionId", "tax_version_id"],
["FileDeliveryStatusID", "file_delivery_status_id"],
["FileDeliveryStatus", "file_delivery_status"],
["FileDeliveryDateTime", "file_delivery_date_time"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.transaction_id,
self.customer_id,
self.customer_name,
self.user_id,
self.user_name,
self.create_date,
self.ui_source_id,
self.ui_source,
self.transaction_status_id,
self.transaction_status,
self.transaction_status_description,
self.transaction_type_id,
self.transaction_type,
self.transaction_type_description,
self.process_date,
self.processed_date,
self.diagnostic,
self.prefunding_status,
self.prefunding_reason,
self.host_account_id,
self.account_number_internal,
self.account_number_external,
self.product_id,
self.product_name,
self.product_type_id,
self.product_type_name,
self.transaction_amount,
self.currency_code,
self.authorized_by_id,
self.authorized_by,
self.authorized_date,
self.modified_by_id,
self.modified_by,
self.modified_date,
self.recurring_transaction_id,
self.description,
self.template_id,
self.host_result,
self.hydra_product_type_code,
self.host_product_type_code,
self.hydra_product_code,
self.host_product_code,
self.processing_type_id,
self.processing_type,
self.processing_type_description,
self.target_description,
self.parent_id,
self.settlement_id,
self.has_children,
self.is_distributed_funding_child,
self.is_settlement_funding_child,
self.skip_balance_check,
self.cif_internal,
self.account_nick_name,
self.effective_date,
self.bill_payee_id,
self.payee_display_name,
self.payee_name,
self.audit_score_id,
self.tran_score_status,
self.tran_score_weight,
self.login_score_id,
self.login_score_status,
self.login_score_weight,
self.group_id,
self.group_desc,
self.to_host_account_id,
self.to_account_number_internal,
self.to_account_number_external,
self.to_product_name,
self.to_product_type_name,
self.has_case_notes,
self.score_audit_id,
self.authorizing_user_credentials,
self.gam_rights_bit_flag,
self.user_role_id,
self.recipient_count,
self.subsidiary_id,
self.gt_batch_id,
self.recipient_name,
self.intermediary_bank_name,
self.beneficiary_bank_name,
self.message_to_beneficiary,
self.template_name,
self.template_description,
self.oob_auth_required,
self.oob_auth_provided,
self.recipient_account_number,
self.recipient_account_type,
self.principal,
self.interest,
self.usd_equivalent_amount,
self.recipient_identification_number,
self.is_same_day_ach,
self.company_entry_description,
self.purpose_of_wire,
self.ach_class_code,
self.has_wage_garnishment_payments,
self.replaces_transaction_id,
self.replaced_by_transaction_id,
self.is_reversed,
self.reversed_gtid,
self.is_reversible,
self.to_product_id,
self.to_hydra_product_type_code,
self.to_hydra_product_code,
self.to_host_product_type_code,
self.to_host_product_code,
self.is_production,
self.zone_id,
self.zone,
self.is_ach_iat,
self.file_name,
self.gt_flavor_id,
self.gt_flavor_short_name,
self.recurring_master,
self.real_time_payment_id,
self.rtp_thread_id,
self.rtp_outgoing,
self.rtp_request_type_short_name,
self.rtp_vendor_short_name,
self.expiry_date,
self.vendor_status,
self.vendor_status_date_time,
self.service_call_unique_identifier,
self.service_call_conversation_identifier,
self.subsidiary_name,
self.is_large_batch_workflow,
self.is_aggregate,
self.cif_external,
self.host_branch,
self.tax_version_id,
self.file_delivery_status_id,
self.file_delivery_status,
self.file_delivery_date_time,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"transaction_id": int,
"customer_id": int,
"customer_name": str,
"user_id": int,
"user_name": str,
"create_date": str,
"ui_source_id": int,
"ui_source": str,
"transaction_status_id": int,
"transaction_status": str,
"transaction_status_description": str,
"transaction_type_id": int,
"transaction_type": str,
"transaction_type_description": str,
"process_date": str,
"processed_date": str,
"diagnostic": bool,
"prefunding_status": str,
"prefunding_reason": str,
"host_account_id": int,
"account_number_internal": str,
"account_number_external": str,
"product_id": int,
"product_name": str,
"product_type_id": int,
"product_type_name": str,
"transaction_amount": float,
"currency_code": str,
"authorized_by_id": int,
"authorized_by": str,
"authorized_date": str,
"modified_by_id": int,
"modified_by": str,
"modified_date": str,
"recurring_transaction_id": int,
"description": str,
"template_id": int,
"host_result": str,
"hydra_product_type_code": str,
"host_product_type_code": str,
"hydra_product_code": str,
"host_product_code": str,
"processing_type_id": int,
"processing_type": str,
"processing_type_description": str,
"target_description": str,
"parent_id": int,
"settlement_id": int,
"has_children": bool,
"is_distributed_funding_child": bool,
"is_settlement_funding_child": bool,
"skip_balance_check": bool,
"cif_internal": str,
"account_nick_name": str,
"effective_date": str,
"bill_payee_id": int,
"payee_display_name": str,
"payee_name": str,
"audit_score_id": int,
"tran_score_status": int,
"tran_score_weight": int,
"login_score_id": int,
"login_score_status": int,
"login_score_weight": int,
"group_id": int,
"group_desc": str,
"to_host_account_id": int,
"to_account_number_internal": str,
"to_account_number_external": str,
"to_product_name": str,
"to_product_type_name": str,
"has_case_notes": bool,
"score_audit_id": int,
"authorizing_user_credentials": str,
"gam_rights_bit_flag": int,
"user_role_id": int,
"recipient_count": int,
"subsidiary_id": int,
"gt_batch_id": int,
"recipient_name": str,
"intermediary_bank_name": str,
"beneficiary_bank_name": str,
"message_to_beneficiary": str,
"template_name": str,
"template_description": str,
"oob_auth_required": bool,
"oob_auth_provided": bool,
"recipient_account_number": str,
"recipient_account_type": str,
"principal": float,
"interest": float,
"usd_equivalent_amount": float,
"recipient_identification_number": str,
"is_same_day_ach": bool,
"company_entry_description": str,
"purpose_of_wire": str,
"ach_class_code": str,
"has_wage_garnishment_payments": bool,
"replaces_transaction_id": int,
"replaced_by_transaction_id": int,
"is_reversed": bool,
"reversed_gtid": int,
"is_reversible": bool,
"to_product_id": int,
"to_hydra_product_type_code": str,
"to_hydra_product_code": str,
"to_host_product_type_code": str,
"to_host_product_code": str,
"is_production": bool,
"zone_id": int,
"zone": str,
"is_ach_iat": bool,
"file_name": str,
"gt_flavor_id": int,
"gt_flavor_short_name": str,
"recurring_master": bool,
"real_time_payment_id": int,
"rtp_thread_id": int,
"rtp_outgoing": bool,
"rtp_request_type_short_name": str,
"rtp_vendor_short_name": str,
"expiry_date": str,
"vendor_status": str,
"vendor_status_date_time": str,
"service_call_unique_identifier": str,
"service_call_conversation_identifier": str,
"subsidiary_name": str,
"is_large_batch_workflow": bool,
"is_aggregate": bool,
"cif_external": str,
"host_branch": str,
"tax_version_id": str,
"file_delivery_status_id": int,
"file_delivery_status": str,
"file_delivery_date_time": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2GeneratedTransactionView":
"""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 Q2ChangeOfAddressTransactionAccounts:
"""
Represents a row in the Q2_ChangeOfAddressTransactionAccounts table.
"""
transaction_id: int = 0
host_account_id: int = 0
account_number_internal: str = ""
account_number_external: Optional[str] = None
product_id: int = 0
product_name: str = ""
hydra_product_code: str = ""
host_product_code: str = ""
product_type_id: int = 0
hydra_product_type_code: str = ""
host_product_type_code: Optional[str] = None
cif_internal: Optional[str] = None
original_record: Optional["Q2ChangeOfAddressTransactionAccounts"] = field(
default=None, repr=False
)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["TransactionID", "transaction_id"],
["HostAccountID", "host_account_id"],
["AccountNumberInternal", "account_number_internal"],
["AccountNumberExternal", "account_number_external"],
["ProductID", "product_id"],
["ProductName", "product_name"],
["HydraProductCode", "hydra_product_code"],
["HostProductCode", "host_product_code"],
["ProductTypeID", "product_type_id"],
["HydraProductTypeCode", "hydra_product_type_code"],
["HostProductTypeCode", "host_product_type_code"],
["CifInternal", "cif_internal"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.transaction_id,
self.host_account_id,
self.account_number_internal,
self.account_number_external,
self.product_id,
self.product_name,
self.hydra_product_code,
self.host_product_code,
self.product_type_id,
self.hydra_product_type_code,
self.host_product_type_code,
self.cif_internal,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"transaction_id": int,
"host_account_id": int,
"account_number_internal": str,
"account_number_external": str,
"product_id": int,
"product_name": str,
"hydra_product_code": str,
"host_product_code": str,
"product_type_id": int,
"hydra_product_type_code": str,
"host_product_type_code": str,
"cif_internal": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2ChangeOfAddressTransactionAccounts":
"""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 Q2ChangeOfAddressTransactions:
"""
Represents a row in the Q2_ChangeOfAddressTransactions table.
"""
transaction_id: int = 0
street_address1: Optional[str] = None
street_address2: Optional[str] = None
city: Optional[str] = None
state: Optional[str] = None
postal_code: Optional[str] = None
old_street_address1: Optional[str] = None
old_street_address2: Optional[str] = None
old_city: Optional[str] = None
old_state: Optional[str] = None
old_postal_code: Optional[str] = None
old_country_id: Optional[int] = None
old_address_type: Optional[str] = None
home_phone_number: Optional[str] = None
work_phone_number: Optional[str] = None
cell_phone_number: Optional[str] = None
email_address: Optional[str] = None
country_id: Optional[int] = None
country_name: Optional[str] = None
old_country_name: Optional[str] = None
province: Optional[str] = None
old_province: Optional[str] = None
original_record: Optional["Q2ChangeOfAddressTransactions"] = field(
default=None, repr=False
)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["TransactionID", "transaction_id"],
["StreetAddress1", "street_address1"],
["StreetAddress2", "street_address2"],
["City", "city"],
["State", "state"],
["PostalCode", "postal_code"],
["OldStreetAddress1", "old_street_address1"],
["OldStreetAddress2", "old_street_address2"],
["OldCity", "old_city"],
["OldState", "old_state"],
["OldPostalCode", "old_postal_code"],
["OldCountryID", "old_country_id"],
["OldAddressType", "old_address_type"],
["HomePhoneNumber", "home_phone_number"],
["WorkPhoneNumber", "work_phone_number"],
["CellPhoneNumber", "cell_phone_number"],
["EmailAddress", "email_address"],
["CountryID", "country_id"],
["CountryName", "country_name"],
["OldCountryName", "old_country_name"],
["Province", "province"],
["OldProvince", "old_province"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.transaction_id,
self.street_address1,
self.street_address2,
self.city,
self.state,
self.postal_code,
self.old_street_address1,
self.old_street_address2,
self.old_city,
self.old_state,
self.old_postal_code,
self.old_country_id,
self.old_address_type,
self.home_phone_number,
self.work_phone_number,
self.cell_phone_number,
self.email_address,
self.country_id,
self.country_name,
self.old_country_name,
self.province,
self.old_province,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"transaction_id": int,
"street_address1": str,
"street_address2": str,
"city": str,
"state": str,
"postal_code": str,
"old_street_address1": str,
"old_street_address2": str,
"old_city": str,
"old_state": str,
"old_postal_code": str,
"old_country_id": int,
"old_address_type": str,
"home_phone_number": str,
"work_phone_number": str,
"cell_phone_number": str,
"email_address": str,
"country_id": int,
"country_name": str,
"old_country_name": str,
"province": str,
"old_province": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2ChangeOfAddressTransactions":
"""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 Q2CheckReOrderTransactions:
"""
Represents a row in the Q2_CheckReOrderTransactions table.
"""
transaction_id: Optional[int] = None
starting_check_number: Optional[str] = None
number_of_boxes: Optional[int] = None
check_line1: Optional[str] = None
check_line2: Optional[str] = None
check_line3: Optional[str] = None
check_line4: Optional[str] = None
original_record: Optional["Q2CheckReOrderTransactions"] = field(
default=None, repr=False
)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["TransactionID", "transaction_id"],
["StartingCheckNumber", "starting_check_number"],
["NumberOfBoxes", "number_of_boxes"],
["CheckLine1", "check_line1"],
["CheckLine2", "check_line2"],
["CheckLine3", "check_line3"],
["CheckLine4", "check_line4"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.transaction_id,
self.starting_check_number,
self.number_of_boxes,
self.check_line1,
self.check_line2,
self.check_line3,
self.check_line4,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"transaction_id": int,
"starting_check_number": str,
"number_of_boxes": int,
"check_line1": str,
"check_line2": str,
"check_line3": str,
"check_line4": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2CheckReOrderTransactions":
"""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 Q2FundsTransferTransactions:
"""
Represents a row in the Q2_FundsTransferTransactions table.
"""
transaction_id: int = 0
principal: Optional[float] = None
interest: Optional[float] = None
transfer_type: Optional[int] = None
to_account_id: Optional[int] = None
account_number_internal: str = ""
account_number_external: Optional[str] = None
product_id: int = 0
product_name: str = ""
hydra_product_code: str = ""
host_product_code: str = ""
product_type_id: int = 0
product_type_name: str = ""
hydra_product_type_code: str = ""
host_product_type_code: Optional[str] = None
cif_internal: Optional[str] = None
gt_data: Optional[str] = None
flavor_bit_flag: Optional[int] = None
cif_external: Optional[str] = None
host_branch: Optional[str] = None
is_external_account: bool = False
aba: Optional[str] = None
original_record: Optional["Q2FundsTransferTransactions"] = field(
default=None, repr=False
)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["TransactionID", "transaction_id"],
["Principal", "principal"],
["Interest", "interest"],
["TransferType", "transfer_type"],
["ToAccountID", "to_account_id"],
["AccountNumberInternal", "account_number_internal"],
["AccountNumberExternal", "account_number_external"],
["ProductID", "product_id"],
["ProductName", "product_name"],
["HydraProductCode", "hydra_product_code"],
["HostProductCode", "host_product_code"],
["ProductTypeID", "product_type_id"],
["ProductTypeName", "product_type_name"],
["HydraProductTypeCode", "hydra_product_type_code"],
["HostProductTypeCode", "host_product_type_code"],
["CIFInternal", "cif_internal"],
["GtData", "gt_data"],
["FlavorBitFlag", "flavor_bit_flag"],
["CIFExternal", "cif_external"],
["HostBranch", "host_branch"],
["IsExternalAccount", "is_external_account"],
["ABA", "aba"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.transaction_id,
self.principal,
self.interest,
self.transfer_type,
self.to_account_id,
self.account_number_internal,
self.account_number_external,
self.product_id,
self.product_name,
self.hydra_product_code,
self.host_product_code,
self.product_type_id,
self.product_type_name,
self.hydra_product_type_code,
self.host_product_type_code,
self.cif_internal,
self.gt_data,
self.flavor_bit_flag,
self.cif_external,
self.host_branch,
self.is_external_account,
self.aba,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"transaction_id": int,
"principal": float,
"interest": float,
"transfer_type": int,
"to_account_id": int,
"account_number_internal": str,
"account_number_external": str,
"product_id": int,
"product_name": str,
"hydra_product_code": str,
"host_product_code": str,
"product_type_id": int,
"product_type_name": str,
"hydra_product_type_code": str,
"host_product_type_code": str,
"cif_internal": str,
"gt_data": str,
"flavor_bit_flag": int,
"cif_external": str,
"host_branch": str,
"is_external_account": bool,
"aba": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2FundsTransferTransactions":
"""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 Q2StopPaymentTransactions:
"""
Represents a row in the Q2_StopPaymentTransactions table.
"""
transaction_id: int = 0
check_number: Optional[str] = None
end_check_number: Optional[str] = None
check_date: Optional[str] = None
payee_name: Optional[str] = None
end_date: Optional[str] = None
original_record: Optional["Q2StopPaymentTransactions"] = field(
default=None, repr=False
)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["TransactionID", "transaction_id"],
["CheckNumber", "check_number"],
["EndCheckNumber", "end_check_number"],
["CheckDate", "check_date"],
["PayeeName", "payee_name"],
["EndDate", "end_date"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.transaction_id,
self.check_number,
self.end_check_number,
self.check_date,
self.payee_name,
self.end_date,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"transaction_id": int,
"check_number": str,
"end_check_number": str,
"check_date": str,
"payee_name": str,
"end_date": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2StopPaymentTransactions":
"""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 Q2BillPaymentTransactions:
"""
Represents a row in the Q2_BillPaymentTransactions table.
"""
transaction_id: int = 0
payee_name: str = ""
street_address1: Optional[str] = None
street_address2: Optional[str] = None
city: Optional[str] = None
state: Optional[str] = None
postal_code: Optional[str] = None
email_address: Optional[str] = None
account_number_or_memo: str = ""
pay_as_name: str = ""
pay_as_address1: str = ""
pay_as_address2: Optional[str] = None
pay_as_city: str = ""
pay_as_state: str = ""
pay_as_postal_code: str = ""
is_international: bool = False
province: Optional[str] = None
city_or_area_code: Optional[str] = None
local_number: Optional[str] = None
extension: Optional[str] = None
name: str = ""
payee_type: str = ""
original_record: Optional["Q2BillPaymentTransactions"] = field(
default=None, repr=False
)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["TransactionID", "transaction_id"],
["PayeeName", "payee_name"],
["StreetAddress1", "street_address1"],
["StreetAddress2", "street_address2"],
["City", "city"],
["State", "state"],
["PostalCode", "postal_code"],
["EmailAddress", "email_address"],
["AccountNumberOrMemo", "account_number_or_memo"],
["PayAsName", "pay_as_name"],
["PayAsAddress1", "pay_as_address1"],
["PayAsAddress2", "pay_as_address2"],
["PayAsCity", "pay_as_city"],
["PayAsState", "pay_as_state"],
["PayAsPostalCode", "pay_as_postal_code"],
["IsInternational", "is_international"],
["Province", "province"],
["CityOrAreaCode", "city_or_area_code"],
["LocalNumber", "local_number"],
["Extension", "extension"],
["Name", "name"],
["PayeeType", "payee_type"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.transaction_id,
self.payee_name,
self.street_address1,
self.street_address2,
self.city,
self.state,
self.postal_code,
self.email_address,
self.account_number_or_memo,
self.pay_as_name,
self.pay_as_address1,
self.pay_as_address2,
self.pay_as_city,
self.pay_as_state,
self.pay_as_postal_code,
self.is_international,
self.province,
self.city_or_area_code,
self.local_number,
self.extension,
self.name,
self.payee_type,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"transaction_id": int,
"payee_name": str,
"street_address1": str,
"street_address2": str,
"city": str,
"state": str,
"postal_code": str,
"email_address": str,
"account_number_or_memo": str,
"pay_as_name": str,
"pay_as_address1": str,
"pay_as_address2": str,
"pay_as_city": str,
"pay_as_state": str,
"pay_as_postal_code": str,
"is_international": bool,
"province": str,
"city_or_area_code": str,
"local_number": str,
"extension": str,
"name": str,
"payee_type": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2BillPaymentTransactions":
"""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 Q2AchPpdCcdTransactions:
"""
Represents a row in the Q2_AchPpdCcdTransactions table.
"""
transaction_id: int = 0
company_name: str = ""
discretionary_data: Optional[str] = None
company_identification: str = ""
ach_class_code_id: int = 0
ach_class_code: str = ""
company_entry_description: str = ""
company_descriptive_date: Optional[str] = None
effective_entry_date: str = ""
originating_tran_code_id: Optional[int] = None
tran_code: Optional[str] = None
tran_code_description: Optional[str] = None
account_type_id: Optional[int] = None
account_type: Optional[str] = None
account_type_description: Optional[str] = None
tran_code_type: Optional[str] = None
addenda: Optional[str] = None
transaction_type_id: int = 0
is_same_day_ach: bool = False
original_record: Optional["Q2AchPpdCcdTransactions"] = field(
default=None, repr=False
)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["TransactionID", "transaction_id"],
["CompanyName", "company_name"],
["DiscretionaryData", "discretionary_data"],
["CompanyIdentification", "company_identification"],
["ACHClassCodeID", "ach_class_code_id"],
["AchClassCode", "ach_class_code"],
["CompanyEntryDescription", "company_entry_description"],
["CompanyDescriptiveDate", "company_descriptive_date"],
["EffectiveEntryDate", "effective_entry_date"],
["OriginatingTranCodeID", "originating_tran_code_id"],
["TranCode", "tran_code"],
["TranCodeDescription", "tran_code_description"],
["AccountTypeID", "account_type_id"],
["AccountType", "account_type"],
["AccountTypeDescription", "account_type_description"],
["TranCodeType", "tran_code_type"],
["Addenda", "addenda"],
["TransactionTypeID", "transaction_type_id"],
["IsSameDayAch", "is_same_day_ach"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.transaction_id,
self.company_name,
self.discretionary_data,
self.company_identification,
self.ach_class_code_id,
self.ach_class_code,
self.company_entry_description,
self.company_descriptive_date,
self.effective_entry_date,
self.originating_tran_code_id,
self.tran_code,
self.tran_code_description,
self.account_type_id,
self.account_type,
self.account_type_description,
self.tran_code_type,
self.addenda,
self.transaction_type_id,
self.is_same_day_ach,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"transaction_id": int,
"company_name": str,
"discretionary_data": str,
"company_identification": str,
"ach_class_code_id": int,
"ach_class_code": str,
"company_entry_description": str,
"company_descriptive_date": str,
"effective_entry_date": str,
"originating_tran_code_id": int,
"tran_code": str,
"tran_code_description": str,
"account_type_id": int,
"account_type": str,
"account_type_description": str,
"tran_code_type": str,
"addenda": str,
"transaction_type_id": int,
"is_same_day_ach": bool,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2AchPpdCcdTransactions":
"""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 Q2AchIatTransactions:
"""
Represents a row in the Q2_AchIatTransactions table.
"""
ach_iat_batch_id: int = 0
transaction_id: int = 0
transaction_type_id: int = 0
originator_identification: str = ""
company_entry_description: str = ""
effective_entry_date: str = ""
recipient_count: int = 0
service_class_code: str = ""
foreign_exchange_indicator: str = ""
destination_currency_code: str = ""
original_record: Optional["Q2AchIatTransactions"] = field(default=None, repr=False)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["AchIatBatchID", "ach_iat_batch_id"],
["TransactionID", "transaction_id"],
["TransactionTypeID", "transaction_type_id"],
["OriginatorIdentification", "originator_identification"],
["CompanyEntryDescription", "company_entry_description"],
["EffectiveEntryDate", "effective_entry_date"],
["RecipientCount", "recipient_count"],
["ServiceClassCode", "service_class_code"],
["ForeignExchangeIndicator", "foreign_exchange_indicator"],
["DestinationCurrencyCode", "destination_currency_code"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.ach_iat_batch_id,
self.transaction_id,
self.transaction_type_id,
self.originator_identification,
self.company_entry_description,
self.effective_entry_date,
self.recipient_count,
self.service_class_code,
self.foreign_exchange_indicator,
self.destination_currency_code,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"ach_iat_batch_id": int,
"transaction_id": int,
"transaction_type_id": int,
"originator_identification": str,
"company_entry_description": str,
"effective_entry_date": str,
"recipient_count": int,
"service_class_code": str,
"foreign_exchange_indicator": str,
"destination_currency_code": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2AchIatTransactions":
"""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 Q2WireDomesticTransactions:
"""
Represents a row in the Q2_WireDomesticTransactions table.
"""
transaction_id: int = 0
wire_transfer_id: int = 0
to_account: str = ""
to_account_type: str = ""
recip_name: Optional[str] = None
recip_address1: Optional[str] = None
recip_address2: Optional[str] = None
recip_city: Optional[str] = None
recip_state: Optional[str] = None
recip_address3: Optional[str] = None
recip_postal_code: Optional[str] = None
recip_country_id: Optional[int] = None
recip_country_name: str = ""
email_address: Optional[str] = None
payer_name: Optional[str] = None
payer_address1: Optional[str] = None
payer_address2: Optional[str] = None
payer_city: Optional[str] = None
payer_state: Optional[str] = None
payer_province: Optional[str] = None
payer_postal: Optional[str] = None
payer_country_id: Optional[int] = None
payer_country_name: Optional[str] = None
rec_fi_aba: Optional[str] = None
rec_fi_name: Optional[str] = None
rec_fi_address1: Optional[str] = None
rec_fi_address2: Optional[str] = None
rec_fi_city: Optional[str] = None
rec_fi_state: Optional[str] = None
rec_fi_postal_code: Optional[str] = None
rec_fi_country_id: Optional[int] = None
rec_fi_country_name: str = ""
intermed_aba: Optional[str] = None
intermed_name: Optional[str] = None
intermed_address1: Optional[str] = None
intermed_address2: Optional[str] = None
intermed_city: Optional[str] = None
intermed_state: Optional[str] = None
intermed_postal_code: Optional[str] = None
intermed_country_id: Optional[int] = None
intermed_country_name: Optional[str] = None
orig_to_benf_info: Optional[str] = None
receiving_fi_short_name: Optional[str] = None
receiving_fi_aba: Optional[str] = None
fi_to_fi_info: Optional[str] = None
ref_for_beneficiary: Optional[str] = None
message_input_reference: Optional[str] = None
imad: Optional[str] = None
purpose_of_wire: Optional[str] = None
swift_code_or_bic: Optional[str] = None
account_num_between_banks: Optional[str] = None
original_record: Optional["Q2WireDomesticTransactions"] = field(
default=None, repr=False
)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["TransactionID", "transaction_id"],
["WireTransferID", "wire_transfer_id"],
["ToAccount", "to_account"],
["ToAccountType", "to_account_type"],
["RecipName", "recip_name"],
["RecipAddress1", "recip_address1"],
["RecipAddress2", "recip_address2"],
["RecipCity", "recip_city"],
["RecipState", "recip_state"],
["RecipAddress3", "recip_address3"],
["RecipPostalCode", "recip_postal_code"],
["RecipCountryId", "recip_country_id"],
["RecipCountryName", "recip_country_name"],
["EmailAddress", "email_address"],
["PayerName", "payer_name"],
["PayerAddress1", "payer_address1"],
["PayerAddress2", "payer_address2"],
["PayerCity", "payer_city"],
["PayerState", "payer_state"],
["PayerProvince", "payer_province"],
["PayerPostal", "payer_postal"],
["PayerCountryID", "payer_country_id"],
["PayerCountryName", "payer_country_name"],
["RecFiABA", "rec_fi_aba"],
["RecFiName", "rec_fi_name"],
["RecFiAddress1", "rec_fi_address1"],
["RecFiAddress2", "rec_fi_address2"],
["RecFiCity", "rec_fi_city"],
["RecFiState", "rec_fi_state"],
["RecFiPostalCode", "rec_fi_postal_code"],
["RecFiCountryID", "rec_fi_country_id"],
["RecFiCountryName", "rec_fi_country_name"],
["IntermedABA", "intermed_aba"],
["IntermedName", "intermed_name"],
["IntermedAddress1", "intermed_address1"],
["IntermedAddress2", "intermed_address2"],
["IntermedCity", "intermed_city"],
["IntermedState", "intermed_state"],
["IntermedPostalCode", "intermed_postal_code"],
["IntermedCountryID", "intermed_country_id"],
["IntermedCountryName", "intermed_country_name"],
["OrigToBenfInfo", "orig_to_benf_info"],
["ReceivingFiShortName", "receiving_fi_short_name"],
["ReceivingFiAba", "receiving_fi_aba"],
["FiToFiInfo", "fi_to_fi_info"],
["RefForBeneficiary", "ref_for_beneficiary"],
["MessageInputReference", "message_input_reference"],
["Imad", "imad"],
["PurposeOfWire", "purpose_of_wire"],
["SwiftCodeOrBic", "swift_code_or_bic"],
["AccountNumBetweenBanks", "account_num_between_banks"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.transaction_id,
self.wire_transfer_id,
self.to_account,
self.to_account_type,
self.recip_name,
self.recip_address1,
self.recip_address2,
self.recip_city,
self.recip_state,
self.recip_address3,
self.recip_postal_code,
self.recip_country_id,
self.recip_country_name,
self.email_address,
self.payer_name,
self.payer_address1,
self.payer_address2,
self.payer_city,
self.payer_state,
self.payer_province,
self.payer_postal,
self.payer_country_id,
self.payer_country_name,
self.rec_fi_aba,
self.rec_fi_name,
self.rec_fi_address1,
self.rec_fi_address2,
self.rec_fi_city,
self.rec_fi_state,
self.rec_fi_postal_code,
self.rec_fi_country_id,
self.rec_fi_country_name,
self.intermed_aba,
self.intermed_name,
self.intermed_address1,
self.intermed_address2,
self.intermed_city,
self.intermed_state,
self.intermed_postal_code,
self.intermed_country_id,
self.intermed_country_name,
self.orig_to_benf_info,
self.receiving_fi_short_name,
self.receiving_fi_aba,
self.fi_to_fi_info,
self.ref_for_beneficiary,
self.message_input_reference,
self.imad,
self.purpose_of_wire,
self.swift_code_or_bic,
self.account_num_between_banks,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"transaction_id": int,
"wire_transfer_id": int,
"to_account": str,
"to_account_type": str,
"recip_name": str,
"recip_address1": str,
"recip_address2": str,
"recip_city": str,
"recip_state": str,
"recip_address3": str,
"recip_postal_code": str,
"recip_country_id": int,
"recip_country_name": str,
"email_address": str,
"payer_name": str,
"payer_address1": str,
"payer_address2": str,
"payer_city": str,
"payer_state": str,
"payer_province": str,
"payer_postal": str,
"payer_country_id": int,
"payer_country_name": str,
"rec_fi_aba": str,
"rec_fi_name": str,
"rec_fi_address1": str,
"rec_fi_address2": str,
"rec_fi_city": str,
"rec_fi_state": str,
"rec_fi_postal_code": str,
"rec_fi_country_id": int,
"rec_fi_country_name": str,
"intermed_aba": str,
"intermed_name": str,
"intermed_address1": str,
"intermed_address2": str,
"intermed_city": str,
"intermed_state": str,
"intermed_postal_code": str,
"intermed_country_id": int,
"intermed_country_name": str,
"orig_to_benf_info": str,
"receiving_fi_short_name": str,
"receiving_fi_aba": str,
"fi_to_fi_info": str,
"ref_for_beneficiary": str,
"message_input_reference": str,
"imad": str,
"purpose_of_wire": str,
"swift_code_or_bic": str,
"account_num_between_banks": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2WireDomesticTransactions":
"""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 Q2ExternalTransferTransactions:
"""
Represents a row in the Q2_ExternalTransferTransactions table.
"""
transaction_id: int = 0
principal: Optional[float] = None
interest: Optional[float] = None
transfer_type: Optional[int] = None
to_account_id: Optional[int] = None
is_same_day_ach: bool = False
account_number_internal: str = ""
account_number_external: Optional[str] = None
product_id: int = 0
product_name: str = ""
hydra_product_code: str = ""
to_host_product_code: str = ""
product_type_id: int = 0
product_type_name: str = ""
hydra_product_type_code: str = ""
to_host_product_type_code: Optional[str] = None
host_user: Optional[str] = None
host_pwd: Optional[str] = None
is_from_account_external: bool = False
from_aba: Optional[str] = None
from_cif_internal: Optional[str] = None
from_cif_external: Optional[str] = None
from_host_product_code: str = ""
from_host_product_type_code: Optional[str] = None
from_host_branch: Optional[str] = None
is_to_account_external: bool = False
to_aba: Optional[str] = None
to_cif_internal: Optional[str] = None
to_cif_external: Optional[str] = None
to_host_branch: Optional[str] = None
original_record: Optional["Q2ExternalTransferTransactions"] = field(
default=None, repr=False
)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["TransactionID", "transaction_id"],
["Principal", "principal"],
["Interest", "interest"],
["TransferType", "transfer_type"],
["ToAccountID", "to_account_id"],
["IsSameDayAch", "is_same_day_ach"],
["AccountNumberInternal", "account_number_internal"],
["AccountNumberExternal", "account_number_external"],
["ProductID", "product_id"],
["ProductName", "product_name"],
["HydraProductCode", "hydra_product_code"],
["ToHostProductCode", "to_host_product_code"],
["ProductTypeID", "product_type_id"],
["ProductTypeName", "product_type_name"],
["HydraProductTypeCode", "hydra_product_type_code"],
["ToHostProductTypeCode", "to_host_product_type_code"],
["HostUser", "host_user"],
["HostPwd", "host_pwd"],
["IsFromAccountExternal", "is_from_account_external"],
["FromABA", "from_aba"],
["FromCIFInternal", "from_cif_internal"],
["FromCIFExternal", "from_cif_external"],
["FromHostProductCode", "from_host_product_code"],
["FromHostProductTypeCode", "from_host_product_type_code"],
["FromHostBranch", "from_host_branch"],
["IsToAccountExternal", "is_to_account_external"],
["ToABA", "to_aba"],
["ToCIFInternal", "to_cif_internal"],
["ToCIFExternal", "to_cif_external"],
["ToHostBranch", "to_host_branch"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.transaction_id,
self.principal,
self.interest,
self.transfer_type,
self.to_account_id,
self.is_same_day_ach,
self.account_number_internal,
self.account_number_external,
self.product_id,
self.product_name,
self.hydra_product_code,
self.to_host_product_code,
self.product_type_id,
self.product_type_name,
self.hydra_product_type_code,
self.to_host_product_type_code,
self.host_user,
self.host_pwd,
self.is_from_account_external,
self.from_aba,
self.from_cif_internal,
self.from_cif_external,
self.from_host_product_code,
self.from_host_product_type_code,
self.from_host_branch,
self.is_to_account_external,
self.to_aba,
self.to_cif_internal,
self.to_cif_external,
self.to_host_branch,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"transaction_id": int,
"principal": float,
"interest": float,
"transfer_type": int,
"to_account_id": int,
"is_same_day_ach": bool,
"account_number_internal": str,
"account_number_external": str,
"product_id": int,
"product_name": str,
"hydra_product_code": str,
"to_host_product_code": str,
"product_type_id": int,
"product_type_name": str,
"hydra_product_type_code": str,
"to_host_product_type_code": str,
"host_user": str,
"host_pwd": str,
"is_from_account_external": bool,
"from_aba": str,
"from_cif_internal": str,
"from_cif_external": str,
"from_host_product_code": str,
"from_host_product_type_code": str,
"from_host_branch": str,
"is_to_account_external": bool,
"to_aba": str,
"to_cif_internal": str,
"to_cif_external": str,
"to_host_branch": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2ExternalTransferTransactions":
"""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 Q2WireInternationalTransactions:
"""
Represents a row in the Q2_WireInternationalTransactions table.
"""
transaction_id: int = 0
wire_transfer_id: int = 0
to_account: str = ""
to_account_type: str = ""
recip_name: Optional[str] = None
recip_address1: Optional[str] = None
recip_address2: Optional[str] = None
recip_address3: Optional[str] = None
recip_email_address: Optional[str] = None
payer_name: Optional[str] = None
payer_address1: Optional[str] = None
payer_address2: Optional[str] = None
payer_city: Optional[str] = None
payer_state: Optional[str] = None
payer_province: Optional[str] = None
payer_postal: Optional[str] = None
payer_country_id: Optional[int] = None
payer_country_name: Optional[str] = None
orig_to_benf_info: Optional[str] = None
rec_fi_swift_or_bic: Optional[str] = None
rec_fi_name: Optional[str] = None
rec_fi_address1: Optional[str] = None
rec_fi_address2: Optional[str] = None
rec_fi_address3: Optional[str] = None
intermed_swift_or_bic: Optional[str] = None
intermed_name: Optional[str] = None
intermed_address1: Optional[str] = None
intermed_address2: Optional[str] = None
intermed_address3: Optional[str] = None
currency_code_id: Optional[int] = None
recip_fi_iban: Optional[str] = None
intermed_fi_iban: Optional[str] = None
recip_country_id: Optional[int] = None
rec_fi_country_id: Optional[int] = None
intermed_country_id: Optional[int] = None
intermed_aba: Optional[str] = None
recip_country_name: Optional[str] = None
rec_fi_country_name: Optional[str] = None
intermed_country_name: Optional[str] = None
quote_id: Optional[str] = None
booking_id: Optional[str] = None
usd_equivalent_amount: Optional[float] = None
exchange_rate: Optional[float] = None
value_date: Optional[str] = None
receiving_fi_short_name: Optional[str] = None
receiving_fi_aba: Optional[str] = None
fi_to_fi_info: Optional[str] = None
ref_for_beneficiary: Optional[str] = None
message_input_reference: Optional[str] = None
imad: Optional[str] = None
purpose_of_wire: Optional[str] = None
rate_applied_to_usd_equivalent: bool = False
user_booking_wire_rate_contract: Optional[str] = None
original_record: Optional["Q2WireInternationalTransactions"] = field(
default=None, repr=False
)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["TransactionID", "transaction_id"],
["WireTransferID", "wire_transfer_id"],
["ToAccount", "to_account"],
["ToAccountType", "to_account_type"],
["RecipName", "recip_name"],
["RecipAddress1", "recip_address1"],
["RecipAddress2", "recip_address2"],
["RecipAddress3", "recip_address3"],
["RecipEmailAddress", "recip_email_address"],
["PayerName", "payer_name"],
["PayerAddress1", "payer_address1"],
["PayerAddress2", "payer_address2"],
["PayerCity", "payer_city"],
["PayerState", "payer_state"],
["PayerProvince", "payer_province"],
["PayerPostal", "payer_postal"],
["PayerCountryID", "payer_country_id"],
["PayerCountryName", "payer_country_name"],
["OrigToBenfInfo", "orig_to_benf_info"],
["RecFiSwiftOrBic", "rec_fi_swift_or_bic"],
["RecFiName", "rec_fi_name"],
["RecFiAddress1", "rec_fi_address1"],
["RecFiAddress2", "rec_fi_address2"],
["RecFiAddress3", "rec_fi_address3"],
["IntermedSwiftOrBic", "intermed_swift_or_bic"],
["IntermedName", "intermed_name"],
["IntermedAddress1", "intermed_address1"],
["IntermedAddress2", "intermed_address2"],
["IntermedAddress3", "intermed_address3"],
["CurrencyCodeID", "currency_code_id"],
["RecipFiIban", "recip_fi_iban"],
["IntermedFiIban", "intermed_fi_iban"],
["RecipCountryID", "recip_country_id"],
["RecFiCountryID", "rec_fi_country_id"],
["IntermedCountryID", "intermed_country_id"],
["IntermedABA", "intermed_aba"],
["RecipCountryName", "recip_country_name"],
["RecFiCountryName", "rec_fi_country_name"],
["IntermedCountryName", "intermed_country_name"],
["QuoteId", "quote_id"],
["BookingId", "booking_id"],
["UsdEquivalentAmount", "usd_equivalent_amount"],
["ExchangeRate", "exchange_rate"],
["ValueDate", "value_date"],
["ReceivingFiShortName", "receiving_fi_short_name"],
["ReceivingFiAba", "receiving_fi_aba"],
["FiToFiInfo", "fi_to_fi_info"],
["RefForBeneficiary", "ref_for_beneficiary"],
["MessageInputReference", "message_input_reference"],
["Imad", "imad"],
["PurposeOfWire", "purpose_of_wire"],
["RateAppliedToUsdEquivalent", "rate_applied_to_usd_equivalent"],
["UserBookingWireRateContract", "user_booking_wire_rate_contract"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.transaction_id,
self.wire_transfer_id,
self.to_account,
self.to_account_type,
self.recip_name,
self.recip_address1,
self.recip_address2,
self.recip_address3,
self.recip_email_address,
self.payer_name,
self.payer_address1,
self.payer_address2,
self.payer_city,
self.payer_state,
self.payer_province,
self.payer_postal,
self.payer_country_id,
self.payer_country_name,
self.orig_to_benf_info,
self.rec_fi_swift_or_bic,
self.rec_fi_name,
self.rec_fi_address1,
self.rec_fi_address2,
self.rec_fi_address3,
self.intermed_swift_or_bic,
self.intermed_name,
self.intermed_address1,
self.intermed_address2,
self.intermed_address3,
self.currency_code_id,
self.recip_fi_iban,
self.intermed_fi_iban,
self.recip_country_id,
self.rec_fi_country_id,
self.intermed_country_id,
self.intermed_aba,
self.recip_country_name,
self.rec_fi_country_name,
self.intermed_country_name,
self.quote_id,
self.booking_id,
self.usd_equivalent_amount,
self.exchange_rate,
self.value_date,
self.receiving_fi_short_name,
self.receiving_fi_aba,
self.fi_to_fi_info,
self.ref_for_beneficiary,
self.message_input_reference,
self.imad,
self.purpose_of_wire,
self.rate_applied_to_usd_equivalent,
self.user_booking_wire_rate_contract,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"transaction_id": int,
"wire_transfer_id": int,
"to_account": str,
"to_account_type": str,
"recip_name": str,
"recip_address1": str,
"recip_address2": str,
"recip_address3": str,
"recip_email_address": str,
"payer_name": str,
"payer_address1": str,
"payer_address2": str,
"payer_city": str,
"payer_state": str,
"payer_province": str,
"payer_postal": str,
"payer_country_id": int,
"payer_country_name": str,
"orig_to_benf_info": str,
"rec_fi_swift_or_bic": str,
"rec_fi_name": str,
"rec_fi_address1": str,
"rec_fi_address2": str,
"rec_fi_address3": str,
"intermed_swift_or_bic": str,
"intermed_name": str,
"intermed_address1": str,
"intermed_address2": str,
"intermed_address3": str,
"currency_code_id": int,
"recip_fi_iban": str,
"intermed_fi_iban": str,
"recip_country_id": int,
"rec_fi_country_id": int,
"intermed_country_id": int,
"intermed_aba": str,
"recip_country_name": str,
"rec_fi_country_name": str,
"intermed_country_name": str,
"quote_id": str,
"booking_id": str,
"usd_equivalent_amount": float,
"exchange_rate": float,
"value_date": str,
"receiving_fi_short_name": str,
"receiving_fi_aba": str,
"fi_to_fi_info": str,
"ref_for_beneficiary": str,
"message_input_reference": str,
"imad": str,
"purpose_of_wire": str,
"rate_applied_to_usd_equivalent": bool,
"user_booking_wire_rate_contract": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2WireInternationalTransactions":
"""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 Q2SendCheckTransactions:
"""
Represents a row in the Q2_SendCheckTransactions table.
"""
transaction_id: int = 0
send_to_acct_owner: bool = False
recipient_name: Optional[str] = None
recipient_address1: Optional[str] = None
recipient_address2: Optional[str] = None
recipient_city: Optional[str] = None
recipient_state: Optional[str] = None
recipient_postal_code: Optional[str] = None
is_cashier_or_certified_check: bool = False
payer_name: Optional[str] = None
payer_address1: Optional[str] = None
payer_address2: Optional[str] = None
payer_city: Optional[str] = None
payer_state: Optional[str] = None
payer_postal_code: Optional[str] = None
original_record: Optional["Q2SendCheckTransactions"] = field(
default=None, repr=False
)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["TransactionID", "transaction_id"],
["SendToAcctOwner", "send_to_acct_owner"],
["RecipientName", "recipient_name"],
["RecipientAddress1", "recipient_address1"],
["RecipientAddress2", "recipient_address2"],
["RecipientCity", "recipient_city"],
["RecipientState", "recipient_state"],
["RecipientPostalCode", "recipient_postal_code"],
["IsCashierOrCertifiedCheck", "is_cashier_or_certified_check"],
["PayerName", "payer_name"],
["PayerAddress1", "payer_address1"],
["PayerAddress2", "payer_address2"],
["PayerCity", "payer_city"],
["PayerState", "payer_state"],
["PayerPostalCode", "payer_postal_code"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.transaction_id,
self.send_to_acct_owner,
self.recipient_name,
self.recipient_address1,
self.recipient_address2,
self.recipient_city,
self.recipient_state,
self.recipient_postal_code,
self.is_cashier_or_certified_check,
self.payer_name,
self.payer_address1,
self.payer_address2,
self.payer_city,
self.payer_state,
self.payer_postal_code,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"transaction_id": int,
"send_to_acct_owner": bool,
"recipient_name": str,
"recipient_address1": str,
"recipient_address2": str,
"recipient_city": str,
"recipient_state": str,
"recipient_postal_code": str,
"is_cashier_or_certified_check": bool,
"payer_name": str,
"payer_address1": str,
"payer_address2": str,
"payer_city": str,
"payer_state": str,
"payer_postal_code": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2SendCheckTransactions":
"""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 Q2AchPassThruTransactions:
"""
Represents a row in the Q2_AchPassThruTransactions table.
"""
ach_pass_thru_id: int = 0
transaction_id: int = 0
debit_total: Optional[float] = None
credit_total: Optional[float] = None
debit_count: Optional[int] = None
credit_count: Optional[int] = None
addenda_count: Optional[int] = None
transaction_type_id: int = 0
batch_count: Optional[int] = None
ach_file: str = ""
file_name: Optional[str] = None
original_record: Optional["Q2AchPassThruTransactions"] = field(
default=None, repr=False
)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["AchPassThruID", "ach_pass_thru_id"],
["TransactionID", "transaction_id"],
["DebitTotal", "debit_total"],
["CreditTotal", "credit_total"],
["DebitCount", "debit_count"],
["CreditCount", "credit_count"],
["AddendaCount", "addenda_count"],
["TransactionTypeID", "transaction_type_id"],
["BatchCount", "batch_count"],
["AchFile", "ach_file"],
["FileName", "file_name"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.ach_pass_thru_id,
self.transaction_id,
self.debit_total,
self.credit_total,
self.debit_count,
self.credit_count,
self.addenda_count,
self.transaction_type_id,
self.batch_count,
self.ach_file,
self.file_name,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"ach_pass_thru_id": int,
"transaction_id": int,
"debit_total": float,
"credit_total": float,
"debit_count": int,
"credit_count": int,
"addenda_count": int,
"transaction_type_id": int,
"batch_count": int,
"ach_file": str,
"file_name": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2AchPassThruTransactions":
"""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 Q2RTPGetTransactions:
"""
Represents a row in the Q2_RTPGetTransactions table.
"""
authorized_date: Optional[str] = None
authorized_user_id: Optional[int] = None
auth_token_id: Optional[int] = None
child_bitflag: int = 0
create_date: str = ""
currency_code_id: Optional[int] = None
currency_code: str = ""
customer_id: int = 0
description: Optional[str] = None
diagnostic: bool = False
account_number_internal: Optional[str] = None
account_number_external: Optional[str] = None
cif_internal: Optional[str] = None
cif_external: Optional[str] = None
host_product_code: str = ""
host_product_type_code: Optional[str] = None
branch_id: Optional[int] = None
gt_credential_type_id: Optional[int] = None
host_result: Optional[str] = None
host_tracking_id: Optional[str] = None
is_reversed: bool = False
modified_date: Optional[str] = None
oob_auth_provided: bool = False
oob_auth_required: bool = False
originating_account_id: Optional[int] = None
parent_id: Optional[int] = None
process_date: Optional[str] = None
processed_date: Optional[str] = None
recurring_master: Optional[bool] = None
recurring_transaction_id: Optional[int] = None
replaced_by_transaction_id: Optional[int] = None
replaces_transaction_id: Optional[int] = None
settlement_id: Optional[int] = None
subsidiary_id_: Optional[int] = None
target_description: Optional[str] = None
template_id: Optional[int] = None
transaction_amount: Optional[float] = None
transaction_id: int = -1
transaction_status_id: int = 0
transaction_type_id: int = 0
ui_source_id: int = 0
user_id: int = 0
user_role_id: Optional[int] = None
real_time_payment_id: int = -1
outgoing: bool = False
local_entity_name: str = ""
local_entity_identifier: Optional[str] = None
local_entity_tax_id: str = ""
local_entity_address1: Optional[str] = None
local_entity_address2: Optional[str] = None
local_entity_address3: Optional[str] = None
local_entity_city: Optional[str] = None
local_entity_state: Optional[str] = None
local_entity_postal_code: Optional[str] = None
local_entity_country_id: Optional[int] = None
local_entity_email_address: Optional[str] = None
local_entity_phone_number: Optional[str] = None
local_entity_phone_country_id: Optional[int] = None
remote_entity_name: str = ""
remote_entity_identifier: Optional[str] = None
remote_entity_address1: Optional[str] = None
remote_entity_address2: Optional[str] = None
remote_entity_address3: Optional[str] = None
remote_entity_city: Optional[str] = None
remote_entity_state: Optional[str] = None
remote_entity_postal_code: Optional[str] = None
remote_entity_country_id: Optional[int] = None
remote_entity_email_address: Optional[str] = None
remote_fi_account_number: str = ""
remote_fi_account_type: Optional[str] = None
remote_fi_routing_number: str = ""
remote_fi_address1: Optional[str] = None
remote_fi_address2: Optional[str] = None
remote_fi_address3: Optional[str] = None
remote_fi_city: Optional[str] = None
remote_fi_state: Optional[str] = None
remote_fi_postal_code: Optional[str] = None
remote_fi_country_id: Optional[int] = None
remote_fi_name: Optional[str] = None
intermed_fi_name: Optional[str] = None
intermed_fi_routing_number: Optional[str] = None
intermed_fi_address1: Optional[str] = None
intermed_fi_address2: Optional[str] = None
intermed_fi_address3: Optional[str] = None
intermed_fi_city: Optional[str] = None
intermed_fi_state: Optional[str] = None
intermed_fi_postal_code: Optional[str] = None
intermed_fi_country_id: Optional[int] = None
request_type_id: int = -1
rtp_request_type_short_name: str = ""
thread_id: int = -1
created_at: str = ""
vendor_id: int = -1
rtp_vendor_short_name: str = ""
end_date: Optional[str] = None
every_x_weeks_or_months: Optional[int] = None
executed_occurrences: Optional[int] = None
first_child_is_same_day: Optional[bool] = None
frequency_bit_flags: Optional[int] = None
last_occurrence: Optional[str] = None
next_occurrence: Optional[str] = None
number_of_occurrences: Optional[int] = None
replaced_by_recurring_transaction_id: Optional[int] = None
replaces_recurring_transaction_id: Optional[int] = None
start_date: Optional[str] = None
recurring_transaction_type: Optional[int] = None
rtp_data_xml: Optional[str] = None
original_record: Optional["Q2RTPGetTransactions"] = field(default=None, repr=False)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["AuthorizedDate", "authorized_date"],
["AuthorizedUserID", "authorized_user_id"],
["AuthTokenID", "auth_token_id"],
["ChildBitflag", "child_bitflag"],
["CreateDate", "create_date"],
["CurrencyCodeID", "currency_code_id"],
["CurrencyCode", "currency_code"],
["CustomerID", "customer_id"],
["Description", "description"],
["Diagnostic", "diagnostic"],
["AccountNumberInternal", "account_number_internal"],
["AccountNumberExternal", "account_number_external"],
["CIFInternal", "cif_internal"],
["CIFExternal", "cif_external"],
["HostProductCode", "host_product_code"],
["HostProductTypeCode", "host_product_type_code"],
["BranchID", "branch_id"],
["GtCredentialTypeID", "gt_credential_type_id"],
["HostResult", "host_result"],
["HostTrackingId", "host_tracking_id"],
["IsReversed", "is_reversed"],
["ModifiedDate", "modified_date"],
["OobAuthProvided", "oob_auth_provided"],
["OobAuthRequired", "oob_auth_required"],
["OriginatingAccountID", "originating_account_id"],
["ParentID", "parent_id"],
["ProcessDate", "process_date"],
["ProcessedDate", "processed_date"],
["RecurringMaster", "recurring_master"],
["RecurringTransactionID", "recurring_transaction_id"],
["ReplacedByTransactionID", "replaced_by_transaction_id"],
["ReplacesTransactionID", "replaces_transaction_id"],
["SettlementID", "settlement_id"],
["SubsidiaryID_", "subsidiary_id_"],
["TargetDescription", "target_description"],
["TemplateID", "template_id"],
["TransactionAmount", "transaction_amount"],
["TransactionID", "transaction_id"],
["TransactionStatusID", "transaction_status_id"],
["TransactionTypeID", "transaction_type_id"],
["UISourceID", "ui_source_id"],
["UserID", "user_id"],
["UserRoleID", "user_role_id"],
["RealTimePaymentID", "real_time_payment_id"],
["Outgoing", "outgoing"],
["LocalEntityName", "local_entity_name"],
["LocalEntityIdentifier", "local_entity_identifier"],
["LocalEntityTaxId", "local_entity_tax_id"],
["LocalEntityAddress1", "local_entity_address1"],
["LocalEntityAddress2", "local_entity_address2"],
["LocalEntityAddress3", "local_entity_address3"],
["LocalEntityCity", "local_entity_city"],
["LocalEntityState", "local_entity_state"],
["LocalEntityPostalCode", "local_entity_postal_code"],
["LocalEntityCountryID", "local_entity_country_id"],
["LocalEntityEmailAddress", "local_entity_email_address"],
["LocalEntityPhoneNumber", "local_entity_phone_number"],
["LocalEntityPhoneCountryID", "local_entity_phone_country_id"],
["RemoteEntityName", "remote_entity_name"],
["RemoteEntityIdentifier", "remote_entity_identifier"],
["RemoteEntityAddress1", "remote_entity_address1"],
["RemoteEntityAddress2", "remote_entity_address2"],
["RemoteEntityAddress3", "remote_entity_address3"],
["RemoteEntityCity", "remote_entity_city"],
["RemoteEntityState", "remote_entity_state"],
["RemoteEntityPostalCode", "remote_entity_postal_code"],
["RemoteEntityCountryID", "remote_entity_country_id"],
["RemoteEntityEmailAddress", "remote_entity_email_address"],
["RemoteFIAccountNumber", "remote_fi_account_number"],
["RemoteFIAccountType", "remote_fi_account_type"],
["RemoteFIRoutingNumber", "remote_fi_routing_number"],
["RemoteFIAddress1", "remote_fi_address1"],
["RemoteFIAddress2", "remote_fi_address2"],
["RemoteFIAddress3", "remote_fi_address3"],
["RemoteFICity", "remote_fi_city"],
["RemoteFIState", "remote_fi_state"],
["RemoteFIPostalCode", "remote_fi_postal_code"],
["RemoteFICountryID", "remote_fi_country_id"],
["RemoteFIName", "remote_fi_name"],
["IntermedFIName", "intermed_fi_name"],
["IntermedFIRoutingNumber", "intermed_fi_routing_number"],
["IntermedFIAddress1", "intermed_fi_address1"],
["IntermedFIAddress2", "intermed_fi_address2"],
["IntermedFIAddress3", "intermed_fi_address3"],
["IntermedFICity", "intermed_fi_city"],
["IntermedFIState", "intermed_fi_state"],
["IntermedFIPostalCode", "intermed_fi_postal_code"],
["IntermedFICountryID", "intermed_fi_country_id"],
["RequestTypeID", "request_type_id"],
["RTPRequestTypeShortName", "rtp_request_type_short_name"],
["ThreadID", "thread_id"],
["CreatedAt", "created_at"],
["VendorID", "vendor_id"],
["RTPVendorShortName", "rtp_vendor_short_name"],
["EndDate", "end_date"],
["EveryXWeeksOrMonths", "every_x_weeks_or_months"],
["ExecutedOccurrences", "executed_occurrences"],
["FirstChildIsSameDay", "first_child_is_same_day"],
["FrequencyBitFlags", "frequency_bit_flags"],
["LastOccurrence", "last_occurrence"],
["NextOccurrence", "next_occurrence"],
["NumberOfOccurrences", "number_of_occurrences"],
[
"ReplacedByRecurringTransactionID",
"replaced_by_recurring_transaction_id",
],
["ReplacesRecurringTransactionID", "replaces_recurring_transaction_id"],
["StartDate", "start_date"],
["RecurringTransactionType", "recurring_transaction_type"],
["RTPDataXml", "rtp_data_xml"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.authorized_date,
self.authorized_user_id,
self.auth_token_id,
self.child_bitflag,
self.create_date,
self.currency_code_id,
self.currency_code,
self.customer_id,
self.description,
self.diagnostic,
self.account_number_internal,
self.account_number_external,
self.cif_internal,
self.cif_external,
self.host_product_code,
self.host_product_type_code,
self.branch_id,
self.gt_credential_type_id,
self.host_result,
self.host_tracking_id,
self.is_reversed,
self.modified_date,
self.oob_auth_provided,
self.oob_auth_required,
self.originating_account_id,
self.parent_id,
self.process_date,
self.processed_date,
self.recurring_master,
self.recurring_transaction_id,
self.replaced_by_transaction_id,
self.replaces_transaction_id,
self.settlement_id,
self.subsidiary_id_,
self.target_description,
self.template_id,
self.transaction_amount,
self.transaction_id,
self.transaction_status_id,
self.transaction_type_id,
self.ui_source_id,
self.user_id,
self.user_role_id,
self.real_time_payment_id,
self.outgoing,
self.local_entity_name,
self.local_entity_identifier,
self.local_entity_tax_id,
self.local_entity_address1,
self.local_entity_address2,
self.local_entity_address3,
self.local_entity_city,
self.local_entity_state,
self.local_entity_postal_code,
self.local_entity_country_id,
self.local_entity_email_address,
self.local_entity_phone_number,
self.local_entity_phone_country_id,
self.remote_entity_name,
self.remote_entity_identifier,
self.remote_entity_address1,
self.remote_entity_address2,
self.remote_entity_address3,
self.remote_entity_city,
self.remote_entity_state,
self.remote_entity_postal_code,
self.remote_entity_country_id,
self.remote_entity_email_address,
self.remote_fi_account_number,
self.remote_fi_account_type,
self.remote_fi_routing_number,
self.remote_fi_address1,
self.remote_fi_address2,
self.remote_fi_address3,
self.remote_fi_city,
self.remote_fi_state,
self.remote_fi_postal_code,
self.remote_fi_country_id,
self.remote_fi_name,
self.intermed_fi_name,
self.intermed_fi_routing_number,
self.intermed_fi_address1,
self.intermed_fi_address2,
self.intermed_fi_address3,
self.intermed_fi_city,
self.intermed_fi_state,
self.intermed_fi_postal_code,
self.intermed_fi_country_id,
self.request_type_id,
self.rtp_request_type_short_name,
self.thread_id,
self.created_at,
self.vendor_id,
self.rtp_vendor_short_name,
self.end_date,
self.every_x_weeks_or_months,
self.executed_occurrences,
self.first_child_is_same_day,
self.frequency_bit_flags,
self.last_occurrence,
self.next_occurrence,
self.number_of_occurrences,
self.replaced_by_recurring_transaction_id,
self.replaces_recurring_transaction_id,
self.start_date,
self.recurring_transaction_type,
self.rtp_data_xml,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"authorized_date": str,
"authorized_user_id": int,
"auth_token_id": int,
"child_bitflag": int,
"create_date": str,
"currency_code_id": int,
"currency_code": str,
"customer_id": int,
"description": str,
"diagnostic": bool,
"account_number_internal": str,
"account_number_external": str,
"cif_internal": str,
"cif_external": str,
"host_product_code": str,
"host_product_type_code": str,
"branch_id": int,
"gt_credential_type_id": int,
"host_result": str,
"host_tracking_id": str,
"is_reversed": bool,
"modified_date": str,
"oob_auth_provided": bool,
"oob_auth_required": bool,
"originating_account_id": int,
"parent_id": int,
"process_date": str,
"processed_date": str,
"recurring_master": bool,
"recurring_transaction_id": int,
"replaced_by_transaction_id": int,
"replaces_transaction_id": int,
"settlement_id": int,
"subsidiary_id_": int,
"target_description": str,
"template_id": int,
"transaction_amount": float,
"transaction_id": int,
"transaction_status_id": int,
"transaction_type_id": int,
"ui_source_id": int,
"user_id": int,
"user_role_id": int,
"real_time_payment_id": int,
"outgoing": bool,
"local_entity_name": str,
"local_entity_identifier": str,
"local_entity_tax_id": str,
"local_entity_address1": str,
"local_entity_address2": str,
"local_entity_address3": str,
"local_entity_city": str,
"local_entity_state": str,
"local_entity_postal_code": str,
"local_entity_country_id": int,
"local_entity_email_address": str,
"local_entity_phone_number": str,
"local_entity_phone_country_id": int,
"remote_entity_name": str,
"remote_entity_identifier": str,
"remote_entity_address1": str,
"remote_entity_address2": str,
"remote_entity_address3": str,
"remote_entity_city": str,
"remote_entity_state": str,
"remote_entity_postal_code": str,
"remote_entity_country_id": int,
"remote_entity_email_address": str,
"remote_fi_account_number": str,
"remote_fi_account_type": str,
"remote_fi_routing_number": str,
"remote_fi_address1": str,
"remote_fi_address2": str,
"remote_fi_address3": str,
"remote_fi_city": str,
"remote_fi_state": str,
"remote_fi_postal_code": str,
"remote_fi_country_id": int,
"remote_fi_name": str,
"intermed_fi_name": str,
"intermed_fi_routing_number": str,
"intermed_fi_address1": str,
"intermed_fi_address2": str,
"intermed_fi_address3": str,
"intermed_fi_city": str,
"intermed_fi_state": str,
"intermed_fi_postal_code": str,
"intermed_fi_country_id": int,
"request_type_id": int,
"rtp_request_type_short_name": str,
"thread_id": int,
"created_at": str,
"vendor_id": int,
"rtp_vendor_short_name": str,
"end_date": str,
"every_x_weeks_or_months": int,
"executed_occurrences": int,
"first_child_is_same_day": bool,
"frequency_bit_flags": int,
"last_occurrence": str,
"next_occurrence": str,
"number_of_occurrences": int,
"replaced_by_recurring_transaction_id": int,
"replaces_recurring_transaction_id": int,
"start_date": str,
"recurring_transaction_type": int,
"rtp_data_xml": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2RTPGetTransactions":
"""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_generated_transactions_dataset(
q2_generated_transaction_view: Optional[
Q2GeneratedTransactionView | list[Q2GeneratedTransactionView]
] = None,
) -> DataSetBuilder:
"""
Build the generated_transactions DataSet parameter for ProcessGeneratedTransactions.
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 'generated_transactions' parameter
"""
dataset = DataSetBuilder("DalGeneratedTransactionView")
# Q2_GeneratedTransactionView
q2_generated_transaction_view_cols = [
col[0] for col in Q2GeneratedTransactionView.get_columns()
]
q2_generated_transaction_view_table = dataset.add_table(
"Q2_GeneratedTransactionView", q2_generated_transaction_view_cols
)
_add_rows_to_table(
q2_generated_transaction_view_table, q2_generated_transaction_view
)
return dataset
[docs]
def build_change_of_address_dataset(
q2_change_of_address_transaction_accounts: Optional[
Q2ChangeOfAddressTransactionAccounts
| list[Q2ChangeOfAddressTransactionAccounts]
] = None,
q2_change_of_address_transactions: Optional[
Q2ChangeOfAddressTransactions | list[Q2ChangeOfAddressTransactions]
] = None,
) -> DataSetBuilder:
"""
Build the change_of_address DataSet parameter for ProcessGeneratedTransactions.
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 'change_of_address' parameter
"""
dataset = DataSetBuilder("DalChangeOfAddressTransactions")
# Q2_ChangeOfAddressTransactionAccounts
q2_change_of_address_transaction_accounts_cols = [
col[0] for col in Q2ChangeOfAddressTransactionAccounts.get_columns()
]
q2_change_of_address_transaction_accounts_table = dataset.add_table(
"Q2_ChangeOfAddressTransactionAccounts",
q2_change_of_address_transaction_accounts_cols,
)
_add_rows_to_table(
q2_change_of_address_transaction_accounts_table,
q2_change_of_address_transaction_accounts,
)
# Q2_ChangeOfAddressTransactions
q2_change_of_address_transactions_cols = [
col[0] for col in Q2ChangeOfAddressTransactions.get_columns()
]
q2_change_of_address_transactions_table = dataset.add_table(
"Q2_ChangeOfAddressTransactions", q2_change_of_address_transactions_cols
)
_add_rows_to_table(
q2_change_of_address_transactions_table, q2_change_of_address_transactions
)
return dataset
[docs]
def build_check_reorder_dataset(
q2_check_re_order_transactions: Optional[
Q2CheckReOrderTransactions | list[Q2CheckReOrderTransactions]
] = None,
) -> DataSetBuilder:
"""
Build the check_reorder DataSet parameter for ProcessGeneratedTransactions.
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 'check_reorder' parameter
"""
dataset = DataSetBuilder("DalCheckReOrderTransactions")
# Q2_CheckReOrderTransactions
q2_check_re_order_transactions_cols = [
col[0] for col in Q2CheckReOrderTransactions.get_columns()
]
q2_check_re_order_transactions_table = dataset.add_table(
"Q2_CheckReOrderTransactions", q2_check_re_order_transactions_cols
)
_add_rows_to_table(
q2_check_re_order_transactions_table, q2_check_re_order_transactions
)
return dataset
[docs]
def build_funds_transfer_dataset(
q2_funds_transfer_transactions: Optional[
Q2FundsTransferTransactions | list[Q2FundsTransferTransactions]
] = None,
) -> DataSetBuilder:
"""
Build the funds_transfer DataSet parameter for ProcessGeneratedTransactions.
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 'funds_transfer' parameter
"""
dataset = DataSetBuilder("DalFundsTransferTransactions")
# Q2_FundsTransferTransactions
q2_funds_transfer_transactions_cols = [
col[0] for col in Q2FundsTransferTransactions.get_columns()
]
q2_funds_transfer_transactions_table = dataset.add_table(
"Q2_FundsTransferTransactions", q2_funds_transfer_transactions_cols
)
_add_rows_to_table(
q2_funds_transfer_transactions_table, q2_funds_transfer_transactions
)
return dataset
[docs]
def build_stop_payment_dataset(
q2_stop_payment_transactions: Optional[
Q2StopPaymentTransactions | list[Q2StopPaymentTransactions]
] = None,
) -> DataSetBuilder:
"""
Build the stop_payment DataSet parameter for ProcessGeneratedTransactions.
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 'stop_payment' parameter
"""
dataset = DataSetBuilder("DalStopPaymentTransactions")
# Q2_StopPaymentTransactions
q2_stop_payment_transactions_cols = [
col[0] for col in Q2StopPaymentTransactions.get_columns()
]
q2_stop_payment_transactions_table = dataset.add_table(
"Q2_StopPaymentTransactions", q2_stop_payment_transactions_cols
)
_add_rows_to_table(q2_stop_payment_transactions_table, q2_stop_payment_transactions)
return dataset
[docs]
def build_bill_payment_dataset(
q2_bill_payment_transactions: Optional[
Q2BillPaymentTransactions | list[Q2BillPaymentTransactions]
] = None,
) -> DataSetBuilder:
"""
Build the bill_payment DataSet parameter for ProcessGeneratedTransactions.
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 'bill_payment' parameter
"""
dataset = DataSetBuilder("DalBillPaymentTransactions")
# Q2_BillPaymentTransactions
q2_bill_payment_transactions_cols = [
col[0] for col in Q2BillPaymentTransactions.get_columns()
]
q2_bill_payment_transactions_table = dataset.add_table(
"Q2_BillPaymentTransactions", q2_bill_payment_transactions_cols
)
_add_rows_to_table(q2_bill_payment_transactions_table, q2_bill_payment_transactions)
return dataset
[docs]
def build_ach_payment_dataset(
q2_ach_ppd_ccd_transactions: Optional[
Q2AchPpdCcdTransactions | list[Q2AchPpdCcdTransactions]
] = None,
) -> DataSetBuilder:
"""
Build the ach_payment DataSet parameter for ProcessGeneratedTransactions.
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 'ach_payment' parameter
"""
dataset = DataSetBuilder("DalAchPpdCcdTransactions")
# Q2_AchPpdCcdTransactions
q2_ach_ppd_ccd_transactions_cols = [
col[0] for col in Q2AchPpdCcdTransactions.get_columns()
]
q2_ach_ppd_ccd_transactions_table = dataset.add_table(
"Q2_AchPpdCcdTransactions", q2_ach_ppd_ccd_transactions_cols
)
_add_rows_to_table(q2_ach_ppd_ccd_transactions_table, q2_ach_ppd_ccd_transactions)
return dataset
[docs]
def build_ach_iat_payment_dataset(
q2_ach_iat_transactions: Optional[
Q2AchIatTransactions | list[Q2AchIatTransactions]
] = None,
) -> DataSetBuilder:
"""
Build the ach_iat_payment DataSet parameter for ProcessGeneratedTransactions.
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 'ach_iat_payment' parameter
"""
dataset = DataSetBuilder("DalAchIatTransactions")
# Q2_AchIatTransactions
q2_ach_iat_transactions_cols = [
col[0] for col in Q2AchIatTransactions.get_columns()
]
q2_ach_iat_transactions_table = dataset.add_table(
"Q2_AchIatTransactions", q2_ach_iat_transactions_cols
)
_add_rows_to_table(q2_ach_iat_transactions_table, q2_ach_iat_transactions)
return dataset
[docs]
def build_ach_receipt_dataset(
q2_ach_ppd_ccd_transactions: Optional[
Q2AchPpdCcdTransactions | list[Q2AchPpdCcdTransactions]
] = None,
) -> DataSetBuilder:
"""
Build the ach_receipt DataSet parameter for ProcessGeneratedTransactions.
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 'ach_receipt' parameter
"""
dataset = DataSetBuilder("DalAchPpdCcdTransactions")
# Q2_AchPpdCcdTransactions
q2_ach_ppd_ccd_transactions_cols = [
col[0] for col in Q2AchPpdCcdTransactions.get_columns()
]
q2_ach_ppd_ccd_transactions_table = dataset.add_table(
"Q2_AchPpdCcdTransactions", q2_ach_ppd_ccd_transactions_cols
)
_add_rows_to_table(q2_ach_ppd_ccd_transactions_table, q2_ach_ppd_ccd_transactions)
return dataset
[docs]
def build_eftps_dataset(
q2_ach_ppd_ccd_transactions: Optional[
Q2AchPpdCcdTransactions | list[Q2AchPpdCcdTransactions]
] = None,
) -> DataSetBuilder:
"""
Build the eftps DataSet parameter for ProcessGeneratedTransactions.
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 'eftps' parameter
"""
dataset = DataSetBuilder("DalAchPpdCcdTransactions")
# Q2_AchPpdCcdTransactions
q2_ach_ppd_ccd_transactions_cols = [
col[0] for col in Q2AchPpdCcdTransactions.get_columns()
]
q2_ach_ppd_ccd_transactions_table = dataset.add_table(
"Q2_AchPpdCcdTransactions", q2_ach_ppd_ccd_transactions_cols
)
_add_rows_to_table(q2_ach_ppd_ccd_transactions_table, q2_ach_ppd_ccd_transactions)
return dataset
[docs]
def build_payroll_dataset(
q2_ach_ppd_ccd_transactions: Optional[
Q2AchPpdCcdTransactions | list[Q2AchPpdCcdTransactions]
] = None,
) -> DataSetBuilder:
"""
Build the payroll DataSet parameter for ProcessGeneratedTransactions.
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 'payroll' parameter
"""
dataset = DataSetBuilder("DalAchPpdCcdTransactions")
# Q2_AchPpdCcdTransactions
q2_ach_ppd_ccd_transactions_cols = [
col[0] for col in Q2AchPpdCcdTransactions.get_columns()
]
q2_ach_ppd_ccd_transactions_table = dataset.add_table(
"Q2_AchPpdCcdTransactions", q2_ach_ppd_ccd_transactions_cols
)
_add_rows_to_table(q2_ach_ppd_ccd_transactions_table, q2_ach_ppd_ccd_transactions)
return dataset
[docs]
def build_wire_domestic_dataset(
q2_wire_domestic_transactions: Optional[
Q2WireDomesticTransactions | list[Q2WireDomesticTransactions]
] = None,
) -> DataSetBuilder:
"""
Build the wire_domestic DataSet parameter for ProcessGeneratedTransactions.
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 'wire_domestic' parameter
"""
dataset = DataSetBuilder("DalWireDomesticTransactions")
# Q2_WireDomesticTransactions
q2_wire_domestic_transactions_cols = [
col[0] for col in Q2WireDomesticTransactions.get_columns()
]
q2_wire_domestic_transactions_table = dataset.add_table(
"Q2_WireDomesticTransactions", q2_wire_domestic_transactions_cols
)
_add_rows_to_table(
q2_wire_domestic_transactions_table, q2_wire_domestic_transactions
)
return dataset
[docs]
def build_ach_collections_dataset(
q2_ach_ppd_ccd_transactions: Optional[
Q2AchPpdCcdTransactions | list[Q2AchPpdCcdTransactions]
] = None,
) -> DataSetBuilder:
"""
Build the ach_collections DataSet parameter for ProcessGeneratedTransactions.
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 'ach_collections' parameter
"""
dataset = DataSetBuilder("DalAchPpdCcdTransactions")
# Q2_AchPpdCcdTransactions
q2_ach_ppd_ccd_transactions_cols = [
col[0] for col in Q2AchPpdCcdTransactions.get_columns()
]
q2_ach_ppd_ccd_transactions_table = dataset.add_table(
"Q2_AchPpdCcdTransactions", q2_ach_ppd_ccd_transactions_cols
)
_add_rows_to_table(q2_ach_ppd_ccd_transactions_table, q2_ach_ppd_ccd_transactions)
return dataset
[docs]
def build_ach_batch_dataset(
q2_ach_ppd_ccd_transactions: Optional[
Q2AchPpdCcdTransactions | list[Q2AchPpdCcdTransactions]
] = None,
) -> DataSetBuilder:
"""
Build the ach_batch DataSet parameter for ProcessGeneratedTransactions.
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 'ach_batch' parameter
"""
dataset = DataSetBuilder("DalAchPpdCcdTransactions")
# Q2_AchPpdCcdTransactions
q2_ach_ppd_ccd_transactions_cols = [
col[0] for col in Q2AchPpdCcdTransactions.get_columns()
]
q2_ach_ppd_ccd_transactions_table = dataset.add_table(
"Q2_AchPpdCcdTransactions", q2_ach_ppd_ccd_transactions_cols
)
_add_rows_to_table(q2_ach_ppd_ccd_transactions_table, q2_ach_ppd_ccd_transactions)
return dataset
[docs]
def build_ach_iat_batch_dataset(
q2_ach_iat_transactions: Optional[
Q2AchIatTransactions | list[Q2AchIatTransactions]
] = None,
) -> DataSetBuilder:
"""
Build the ach_iat_batch DataSet parameter for ProcessGeneratedTransactions.
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 'ach_iat_batch' parameter
"""
dataset = DataSetBuilder("DalAchIatTransactions")
# Q2_AchIatTransactions
q2_ach_iat_transactions_cols = [
col[0] for col in Q2AchIatTransactions.get_columns()
]
q2_ach_iat_transactions_table = dataset.add_table(
"Q2_AchIatTransactions", q2_ach_iat_transactions_cols
)
_add_rows_to_table(q2_ach_iat_transactions_table, q2_ach_iat_transactions)
return dataset
[docs]
def build_external_transfer_dataset(
q2_external_transfer_transactions: Optional[
Q2ExternalTransferTransactions | list[Q2ExternalTransferTransactions]
] = None,
) -> DataSetBuilder:
"""
Build the external_transfer DataSet parameter for ProcessGeneratedTransactions.
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 'external_transfer' parameter
"""
dataset = DataSetBuilder("DalExternalTransferTransactions")
# Q2_ExternalTransferTransactions
q2_external_transfer_transactions_cols = [
col[0] for col in Q2ExternalTransferTransactions.get_columns()
]
q2_external_transfer_transactions_table = dataset.add_table(
"Q2_ExternalTransferTransactions", q2_external_transfer_transactions_cols
)
_add_rows_to_table(
q2_external_transfer_transactions_table, q2_external_transfer_transactions
)
return dataset
[docs]
def build_wire_international_dataset(
q2_wire_international_transactions: Optional[
Q2WireInternationalTransactions | list[Q2WireInternationalTransactions]
] = None,
) -> DataSetBuilder:
"""
Build the wire_international DataSet parameter for ProcessGeneratedTransactions.
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 'wire_international' parameter
"""
dataset = DataSetBuilder("DalWireInternationalTransactions")
# Q2_WireInternationalTransactions
q2_wire_international_transactions_cols = [
col[0] for col in Q2WireInternationalTransactions.get_columns()
]
q2_wire_international_transactions_table = dataset.add_table(
"Q2_WireInternationalTransactions", q2_wire_international_transactions_cols
)
_add_rows_to_table(
q2_wire_international_transactions_table, q2_wire_international_transactions
)
return dataset
[docs]
def build_send_check_dataset(
q2_send_check_transactions: Optional[
Q2SendCheckTransactions | list[Q2SendCheckTransactions]
] = None,
) -> DataSetBuilder:
"""
Build the send_check DataSet parameter for ProcessGeneratedTransactions.
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 'send_check' parameter
"""
dataset = DataSetBuilder("DalSendCheckTransactions")
# Q2_SendCheckTransactions
q2_send_check_transactions_cols = [
col[0] for col in Q2SendCheckTransactions.get_columns()
]
q2_send_check_transactions_table = dataset.add_table(
"Q2_SendCheckTransactions", q2_send_check_transactions_cols
)
_add_rows_to_table(q2_send_check_transactions_table, q2_send_check_transactions)
return dataset
[docs]
def build_ach_pass_thru_dataset(
q2_ach_pass_thru_transactions: Optional[
Q2AchPassThruTransactions | list[Q2AchPassThruTransactions]
] = None,
) -> DataSetBuilder:
"""
Build the ach_pass_thru DataSet parameter for ProcessGeneratedTransactions.
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 'ach_pass_thru' parameter
"""
dataset = DataSetBuilder("DalAchPassThruTransactions")
# Q2_AchPassThruTransactions
q2_ach_pass_thru_transactions_cols = [
col[0] for col in Q2AchPassThruTransactions.get_columns()
]
q2_ach_pass_thru_transactions_table = dataset.add_table(
"Q2_AchPassThruTransactions", q2_ach_pass_thru_transactions_cols
)
_add_rows_to_table(
q2_ach_pass_thru_transactions_table, q2_ach_pass_thru_transactions
)
return dataset
[docs]
def build_rtp_credit_transfer_dataset(
q2_rtp_get_transactions: Optional[
Q2RTPGetTransactions | list[Q2RTPGetTransactions]
] = None,
) -> DataSetBuilder:
"""
Build the rtp_credit_transfer DataSet parameter for ProcessGeneratedTransactions.
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 'rtp_credit_transfer' parameter
"""
dataset = DataSetBuilder("DalRTPGetTransactions")
# Q2_RTPGetTransactions
q2_rtp_get_transactions_cols = [
col[0] for col in Q2RTPGetTransactions.get_columns()
]
q2_rtp_get_transactions_table = dataset.add_table(
"Q2_RTPGetTransactions", q2_rtp_get_transactions_cols
)
_add_rows_to_table(q2_rtp_get_transactions_table, q2_rtp_get_transactions)
return dataset
[docs]
def build_rtp_request_for_payment_dataset(
q2_rtp_get_transactions: Optional[
Q2RTPGetTransactions | list[Q2RTPGetTransactions]
] = None,
) -> DataSetBuilder:
"""
Build the rtp_request_for_payment DataSet parameter for ProcessGeneratedTransactions.
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 'rtp_request_for_payment' parameter
"""
dataset = DataSetBuilder("DalRTPGetTransactions")
# Q2_RTPGetTransactions
q2_rtp_get_transactions_cols = [
col[0] for col in Q2RTPGetTransactions.get_columns()
]
q2_rtp_get_transactions_table = dataset.add_table(
"Q2_RTPGetTransactions", q2_rtp_get_transactions_cols
)
_add_rows_to_table(q2_rtp_get_transactions_table, q2_rtp_get_transactions)
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)