# This is an autogenerated file from the command "q2 generate_hq_api" and will be overwritten if run again
"""
DataSet models for CsrSetPolicyDataByPolicyIdentifier.
These typed dataclass models represent the tables and columns within the
DataSet parameters for the CsrSetPolicyDataByPolicyIdentifier 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 CsrSetPolicyDataByPolicyIdentifier
q2_policy_generated_transaction_rights = CsrSetPolicyDataByPolicyIdentifier.Q2PolicyGeneratedTransactionRights(
policy_identifier="...",
short_name="...",
)
q2_policy_subsidiaries = CsrSetPolicyDataByPolicyIdentifier.Q2PolicySubsidiaries(
)
q2_policy_data = CsrSetPolicyDataByPolicyIdentifier.Q2PolicyData(
policy_identifier="...",
description="...",
)
q2_policy_object_groups = CsrSetPolicyDataByPolicyIdentifier.Q2PolicyObjectGroups(
short_name="...",
data="...",
alias="...",
object_group_type="...",
)
q2_policy_accounts = CsrSetPolicyDataByPolicyIdentifier.Q2PolicyAccounts(
customer_account_id=1,
host_account_id=1,
customer_access=1,
account_number_internal="...",
status_description="...",
product_type_name="...",
product_name="...",
is_external_account=True,
product_type_id=1,
product_id=1,
)
q2_policy_features = CsrSetPolicyDataByPolicyIdentifier.Q2PolicyFeatures(
policy_identifier="...",
enabled=True,
property_name="...",
)
q2_policy_ach_class_codes = CsrSetPolicyDataByPolicyIdentifier.Q2PolicyAchClassCodes(
ach_class_code="...",
)
"""
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 Q2PolicyGeneratedTransactionRights:
"""
Represents a row in the Q2_PolicyGeneratedTransactionRights table.
"""
policy_identifier: str = ""
transaction_type_id: Optional[int] = None
short_name: str = ""
group_id: Optional[int] = None
customer_id: Optional[int] = None
company_id: Optional[int] = None
user_role_id: Optional[int] = None
user_id: Optional[int] = None
enabled: Optional[bool] = None
authorize: Optional[bool] = None
cancel: Optional[bool] = None
create: Optional[bool] = None
create_restricted: Optional[bool] = None
update: Optional[bool] = None
view: Optional[int] = None
limit_per_transaction: Optional[float] = None
limit_per_day: Optional[float] = None
limit_per_month: Optional[float] = None
limit_per_acct_per_day: Optional[float] = None
count_per_day: Optional[int] = None
count_per_month: Optional[int] = None
count_per_acct_per_day: Optional[int] = None
dual_approval_limit: Optional[float] = None
token_required_limit: Optional[float] = None
draft_limit: Optional[float] = None
rule_data: Optional[str] = None
source_accounts: Optional[str] = None
destination_accounts: Optional[str] = None
original_record: Optional["Q2PolicyGeneratedTransactionRights"] = field(
default=None, repr=False
)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["PolicyIdentifier", "policy_identifier"],
["TransactionTypeID", "transaction_type_id"],
["ShortName", "short_name"],
["GroupID", "group_id"],
["CustomerID", "customer_id"],
["CompanyID", "company_id"],
["UserRoleID", "user_role_id"],
["UserID", "user_id"],
["Enabled", "enabled"],
["Authorize", "authorize"],
["Cancel", "cancel"],
["Create", "create"],
["CreateRestricted", "create_restricted"],
["Update", "update"],
["View", "view"],
["LimitPerTransaction", "limit_per_transaction"],
["LimitPerDay", "limit_per_day"],
["LimitPerMonth", "limit_per_month"],
["LimitPerAcctPerDay", "limit_per_acct_per_day"],
["CountPerDay", "count_per_day"],
["CountPerMonth", "count_per_month"],
["CountPerAcctPerDay", "count_per_acct_per_day"],
["DualApprovalLimit", "dual_approval_limit"],
["TokenRequiredLimit", "token_required_limit"],
["DraftLimit", "draft_limit"],
["RuleData", "rule_data"],
["SourceAccounts", "source_accounts"],
["DestinationAccounts", "destination_accounts"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.policy_identifier,
self.transaction_type_id,
self.short_name,
self.group_id,
self.customer_id,
self.company_id,
self.user_role_id,
self.user_id,
self.enabled,
self.authorize,
self.cancel,
self.create,
self.create_restricted,
self.update,
self.view,
self.limit_per_transaction,
self.limit_per_day,
self.limit_per_month,
self.limit_per_acct_per_day,
self.count_per_day,
self.count_per_month,
self.count_per_acct_per_day,
self.dual_approval_limit,
self.token_required_limit,
self.draft_limit,
self.rule_data,
self.source_accounts,
self.destination_accounts,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"policy_identifier": str,
"transaction_type_id": int,
"short_name": str,
"group_id": int,
"customer_id": int,
"company_id": int,
"user_role_id": int,
"user_id": int,
"enabled": bool,
"authorize": bool,
"cancel": bool,
"create": bool,
"create_restricted": bool,
"update": bool,
"view": int,
"limit_per_transaction": float,
"limit_per_day": float,
"limit_per_month": float,
"limit_per_acct_per_day": float,
"count_per_day": int,
"count_per_month": int,
"count_per_acct_per_day": int,
"dual_approval_limit": float,
"token_required_limit": float,
"draft_limit": float,
"rule_data": str,
"source_accounts": str,
"destination_accounts": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2PolicyGeneratedTransactionRights":
"""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 Q2PolicySubsidiaries:
"""
Represents a row in the Q2_PolicySubsidiaries table.
"""
policy_identifier: Optional[str] = None
subsidiary_id: Optional[int] = None
name: Optional[str] = None
original_record: Optional["Q2PolicySubsidiaries"] = field(default=None, repr=False)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["PolicyIdentifier", "policy_identifier"],
["SubsidiaryID", "subsidiary_id"],
["Name", "name"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.policy_identifier,
self.subsidiary_id,
self.name,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"policy_identifier": str,
"subsidiary_id": int,
"name": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2PolicySubsidiaries":
"""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 Q2PolicyData:
"""
Represents a row in the Q2_PolicyData table.
"""
policy_identifier: str = ""
policy_data_type: Optional[str] = None
policy_type_id: Optional[int] = None
description: str = ""
has_pending_change: Optional[bool] = None
rule_data: Optional[str] = None
source_accounts: Optional[str] = None
destination_accounts: Optional[str] = None
create_date: Optional[str] = None
last_edit_date: Optional[str] = None
enabled: Optional[bool] = None
is_treasury: Optional[bool] = None
parent_policy_identifier: Optional[str] = None
extended_description: Optional[str] = None
is_template: Optional[bool] = None
is_open_banking: Optional[bool] = None
original_record: Optional["Q2PolicyData"] = field(default=None, repr=False)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["PolicyIdentifier", "policy_identifier"],
["PolicyDataType", "policy_data_type"],
["PolicyTypeId", "policy_type_id"],
["Description", "description"],
["HasPendingChange", "has_pending_change"],
["RuleData", "rule_data"],
["SourceAccounts", "source_accounts"],
["DestinationAccounts", "destination_accounts"],
["CreateDate", "create_date"],
["LastEditDate", "last_edit_date"],
["Enabled", "enabled"],
["IsTreasury", "is_treasury"],
["ParentPolicyIdentifier", "parent_policy_identifier"],
["ExtendedDescription", "extended_description"],
["IsTemplate", "is_template"],
["IsOpenBanking", "is_open_banking"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.policy_identifier,
self.policy_data_type,
self.policy_type_id,
self.description,
self.has_pending_change,
self.rule_data,
self.source_accounts,
self.destination_accounts,
self.create_date,
self.last_edit_date,
self.enabled,
self.is_treasury,
self.parent_policy_identifier,
self.extended_description,
self.is_template,
self.is_open_banking,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"policy_identifier": str,
"policy_data_type": str,
"policy_type_id": int,
"description": str,
"has_pending_change": bool,
"rule_data": str,
"source_accounts": str,
"destination_accounts": str,
"create_date": str,
"last_edit_date": str,
"enabled": bool,
"is_treasury": bool,
"parent_policy_identifier": str,
"extended_description": str,
"is_template": bool,
"is_open_banking": bool,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2PolicyData":
"""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 Q2PolicyObjectGroups:
"""
Represents a row in the Q2_PolicyObjectGroups table.
"""
policy_identifier: Optional[str] = None
object_group_id: int = -1
short_name: str = ""
customer_id: Optional[int] = None
data: str = ""
alias: str = ""
object_group_type: str = ""
group_type_description: Optional[str] = None
description: Optional[str] = None
user_id: Optional[int] = None
original_record: Optional["Q2PolicyObjectGroups"] = field(default=None, repr=False)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["PolicyIdentifier", "policy_identifier"],
["ObjectGroupID", "object_group_id"],
["ShortName", "short_name"],
["CustomerID", "customer_id"],
["Data", "data"],
["Alias", "alias"],
["ObjectGroupType", "object_group_type"],
["GroupTypeDescription", "group_type_description"],
["Description", "description"],
["UserID", "user_id"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.policy_identifier,
self.object_group_id,
self.short_name,
self.customer_id,
self.data,
self.alias,
self.object_group_type,
self.group_type_description,
self.description,
self.user_id,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"policy_identifier": str,
"object_group_id": int,
"short_name": str,
"customer_id": int,
"data": str,
"alias": str,
"object_group_type": str,
"group_type_description": str,
"description": str,
"user_id": int,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2PolicyObjectGroups":
"""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 Q2PolicyAccounts:
"""
Represents a row in the Q2_PolicyAccounts table.
"""
policy_identifier: Optional[str] = None
customer_account_id: int = 0
host_account_id: int = 0
customer_access: int = 0
policy_reference_id: Optional[int] = None
policy_access: Optional[int] = None
account_number_internal: str = ""
account_number_external: Optional[str] = None
account_desc: Optional[str] = None
status_description: str = ""
product_type_name: str = ""
product_name: str = ""
aba: Optional[str] = None
is_external_account: bool = False
account_label: Optional[str] = None
product_type_id: int = 0
product_id: int = 0
is_cdis_enabled: Optional[bool] = None
original_record: Optional["Q2PolicyAccounts"] = field(default=None, repr=False)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["PolicyIdentifier", "policy_identifier"],
["CustomerAccountID", "customer_account_id"],
["HostAccountID", "host_account_id"],
["CustomerAccess", "customer_access"],
["PolicyReferenceID", "policy_reference_id"],
["PolicyAccess", "policy_access"],
["AccountNumberInternal", "account_number_internal"],
["AccountNumberExternal", "account_number_external"],
["AccountDesc", "account_desc"],
["StatusDescription", "status_description"],
["ProductTypeName", "product_type_name"],
["ProductName", "product_name"],
["ABA", "aba"],
["IsExternalAccount", "is_external_account"],
["AccountLabel", "account_label"],
["ProductTypeID", "product_type_id"],
["ProductID", "product_id"],
["IsCdisEnabled", "is_cdis_enabled"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.policy_identifier,
self.customer_account_id,
self.host_account_id,
self.customer_access,
self.policy_reference_id,
self.policy_access,
self.account_number_internal,
self.account_number_external,
self.account_desc,
self.status_description,
self.product_type_name,
self.product_name,
self.aba,
self.is_external_account,
self.account_label,
self.product_type_id,
self.product_id,
self.is_cdis_enabled,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"policy_identifier": str,
"customer_account_id": int,
"host_account_id": int,
"customer_access": int,
"policy_reference_id": int,
"policy_access": int,
"account_number_internal": str,
"account_number_external": str,
"account_desc": str,
"status_description": str,
"product_type_name": str,
"product_name": str,
"aba": str,
"is_external_account": bool,
"account_label": str,
"product_type_id": int,
"product_id": int,
"is_cdis_enabled": bool,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2PolicyAccounts":
"""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 Q2PolicyFeatures:
"""
Represents a row in the Q2_PolicyFeatures table.
"""
policy_identifier: str = ""
user_property_data_id: Optional[int] = None
group_id: Optional[int] = None
customer_id: Optional[int] = None
user_id: Optional[int] = None
user_role_id: Optional[int] = None
property_id: Optional[int] = None
enabled: bool = False
property_name: str = ""
category_short_name: Optional[str] = None
category_description: Optional[str] = None
allow_view: Optional[bool] = None
allow_edit: Optional[bool] = None
original_record: Optional["Q2PolicyFeatures"] = field(default=None, repr=False)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["PolicyIdentifier", "policy_identifier"],
["UserPropertyDataID", "user_property_data_id"],
["GroupID", "group_id"],
["CustomerID", "customer_id"],
["UserID", "user_id"],
["UserRoleID", "user_role_id"],
["PropertyID", "property_id"],
["Enabled", "enabled"],
["PropertyName", "property_name"],
["CategoryShortName", "category_short_name"],
["CategoryDescription", "category_description"],
["AllowView", "allow_view"],
["AllowEdit", "allow_edit"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.policy_identifier,
self.user_property_data_id,
self.group_id,
self.customer_id,
self.user_id,
self.user_role_id,
self.property_id,
self.enabled,
self.property_name,
self.category_short_name,
self.category_description,
self.allow_view,
self.allow_edit,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"policy_identifier": str,
"user_property_data_id": int,
"group_id": int,
"customer_id": int,
"user_id": int,
"user_role_id": int,
"property_id": int,
"enabled": bool,
"property_name": str,
"category_short_name": str,
"category_description": str,
"allow_view": bool,
"allow_edit": bool,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2PolicyFeatures":
"""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 Q2PolicyAchClassCodes:
"""
Represents a row in the Q2_PolicyAchClassCodes table.
"""
policy_identifier: Optional[str] = None
ach_class_code: str = ""
original_record: Optional["Q2PolicyAchClassCodes"] = field(default=None, repr=False)
[docs]
@staticmethod
def get_columns() -> list[list[str]]:
"""Returns [[HqColumnName, pythonAttrName], ...] mapping."""
return [
["PolicyIdentifier", "policy_identifier"],
["AchClassCode", "ach_class_code"],
]
[docs]
def to_row_values(self) -> list[Any]:
"""Returns column values in get_columns() order."""
return [
self.policy_identifier,
self.ach_class_code,
]
_COLUMN_TYPES: ClassVar[dict[str, type]] = {
"policy_identifier": str,
"ach_class_code": str,
}
[docs]
@classmethod
def from_hq_response(cls, row_elem) -> "Q2PolicyAchClassCodes":
"""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_policy_data_dataset(
q2_policy_generated_transaction_rights: Optional[
Q2PolicyGeneratedTransactionRights | list[Q2PolicyGeneratedTransactionRights]
] = None,
q2_policy_subsidiaries: Optional[
Q2PolicySubsidiaries | list[Q2PolicySubsidiaries]
] = None,
q2_policy_data: Optional[Q2PolicyData | list[Q2PolicyData]] = None,
q2_policy_object_groups: Optional[
Q2PolicyObjectGroups | list[Q2PolicyObjectGroups]
] = None,
q2_policy_accounts: Optional[Q2PolicyAccounts | list[Q2PolicyAccounts]] = None,
q2_policy_features: Optional[Q2PolicyFeatures | list[Q2PolicyFeatures]] = None,
q2_policy_ach_class_codes: Optional[
Q2PolicyAchClassCodes | list[Q2PolicyAchClassCodes]
] = None,
) -> DataSetBuilder:
"""
Build the policy_data DataSet parameter for CsrSetPolicyDataByPolicyIdentifier.
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 'policy_data' parameter
"""
dataset = DataSetBuilder("DalPolicyData")
# Q2_PolicyGeneratedTransactionRights
q2_policy_generated_transaction_rights_cols = [
col[0] for col in Q2PolicyGeneratedTransactionRights.get_columns()
]
q2_policy_generated_transaction_rights_table = dataset.add_table(
"Q2_PolicyGeneratedTransactionRights",
q2_policy_generated_transaction_rights_cols,
)
_add_rows_to_table(
q2_policy_generated_transaction_rights_table,
q2_policy_generated_transaction_rights,
)
# Q2_PolicySubsidiaries
q2_policy_subsidiaries_cols = [col[0] for col in Q2PolicySubsidiaries.get_columns()]
q2_policy_subsidiaries_table = dataset.add_table(
"Q2_PolicySubsidiaries", q2_policy_subsidiaries_cols
)
_add_rows_to_table(q2_policy_subsidiaries_table, q2_policy_subsidiaries)
# Q2_PolicyData
q2_policy_data_cols = [col[0] for col in Q2PolicyData.get_columns()]
q2_policy_data_table = dataset.add_table("Q2_PolicyData", q2_policy_data_cols)
_add_rows_to_table(q2_policy_data_table, q2_policy_data)
# Q2_PolicyObjectGroups
q2_policy_object_groups_cols = [
col[0] for col in Q2PolicyObjectGroups.get_columns()
]
q2_policy_object_groups_table = dataset.add_table(
"Q2_PolicyObjectGroups", q2_policy_object_groups_cols
)
_add_rows_to_table(q2_policy_object_groups_table, q2_policy_object_groups)
# Q2_PolicyAccounts
q2_policy_accounts_cols = [col[0] for col in Q2PolicyAccounts.get_columns()]
q2_policy_accounts_table = dataset.add_table(
"Q2_PolicyAccounts", q2_policy_accounts_cols
)
_add_rows_to_table(q2_policy_accounts_table, q2_policy_accounts)
# Q2_PolicyFeatures
q2_policy_features_cols = [col[0] for col in Q2PolicyFeatures.get_columns()]
q2_policy_features_table = dataset.add_table(
"Q2_PolicyFeatures", q2_policy_features_cols
)
_add_rows_to_table(q2_policy_features_table, q2_policy_features)
# Q2_PolicyAchClassCodes
q2_policy_ach_class_codes_cols = [
col[0] for col in Q2PolicyAchClassCodes.get_columns()
]
q2_policy_ach_class_codes_table = dataset.add_table(
"Q2_PolicyAchClassCodes", q2_policy_ach_class_codes_cols
)
_add_rows_to_table(q2_policy_ach_class_codes_table, q2_policy_ach_class_codes)
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)