Source code for q2_sdk.hq.api_helpers

"""
A few of the HQ endpoints have some difficult to document inputs.
For instance, xml_payload shows up in several of them, which can
literally be any shape of xml. However, there is an appropriate
shape per endpoint, which we try to build out here in this file.

You can pass the result of one of these functions as input to the
functions generated into hq_api with generate_hq_api.

For instance:

    from q2_sdk.hq.hq_api.q2_api import AddCustomer

    xml_payload = build_add_customer_xml(demo_info, group_id, is_company)
    AddCustomer.ParamsObj(self.logger, xml_payload)
"""

import base64
from datetime import datetime, timedelta
import random
import string
from typing import Optional

from lxml import etree
from q2_sdk.hq.models.password_policy import PasswordPolicy
from q2_sdk.hq.models.policy_data import (
    Account,
    Company,
    Customer,
    Data,
    Features,
    GeneratedTransactionRights,
    Group,
    PolicyData,
    Subsidiaries,
    User,
    UserRole,
)
from q2_sdk.hq.models.secure_message_attachment import SecureMessageAttachment
from q2_sdk.hq.models.secure_messenger_type import SecureMessengerType
from q2_sdk.hq.models.transaction_info import DayOfWeek, WeekOfMonth
from q2_sdk.models.demographic import (
    Address,
    AddressType,
    DemographicInfo,
    Phone,
    PhoneType,
)
from q2_sdk.models.subsidiary import Subsidiary


[docs] def build_add_customer_xml( demo_info: DemographicInfo, group_id: int, is_company: bool, subsidiaries: Optional[list[Subsidiary]] = None, host_user: Optional[str] = None, host_pwd: Optional[str] = None, allow_other_address_types=False, ) -> str: """ Transforms DemographicInfo into an xml_payload for use with HQ's AddCustomer :param demo_info: DemographicInfo instance :param group_id: Which group the customer will be inserted into :param is_company: Sets some database fields appropriately :param subsidiaries: List of subsidiaries to add to Q2 customer. Normally used with commercial and/or treasury users :param host_user: populates the HostUser column on the Q2_Customer table :param host_pwd: populates the HostPwd column on the Q2_Customer table :param allow_other_address_types: turns off the Home only logic if True :return: Already escaped xml_payload """ root = etree.Element("DalCustomerEdit") q2_customer_node = _build_customer_node( demo_info, group_id, is_company, host_user, host_pwd ) root.append(q2_customer_node) for address in demo_info.addresses: if address.address_type != AddressType.HOME and not allow_other_address_types: continue root.append(_build_address_node(address)) for phone in demo_info.phones: root.append(_build_phone_node(phone)) if subsidiaries: for subsidiary in subsidiaries: root.append(_build_subsidiary_node(subsidiary)) return etree.tostring(root).decode("utf8")
[docs] def build_add_user_xml( demo_info: DemographicInfo, customer_id: int, host_user: Optional[str] = None, host_pwd: Optional[str] = None, allow_other_address_types=False, *, null_primary_cif: bool = False, ) -> str: """ Transforms DemographicInfo into an xml_payload for use with HQ's AddUser :param demo_info: DemographicInfo instance :param customer_id: Which customer this user belongs to :param host_user: populates the HostUser column on the Q2_User table :param host_pwd: populates the HostPwd column on the Q2_User table :param allow_other_address_types: turns off the Home only logic if True :return: Already escaped xml_payload """ root = etree.Element("DalUserEdit") q2_user_node = _build_user_node( demo_info, customer_id, host_user, host_pwd, null_primary_cif=null_primary_cif ) root.append(q2_user_node) for address in demo_info.addresses: if address.address_type != AddressType.HOME and not allow_other_address_types: continue root.append(_build_address_node(address)) for phone in demo_info.phones: sac_target = phone.is_sac_target sms_target = phone.is_sac_sms_target root.append( _build_phone_node(phone, with_access=sac_target, with_sms=sms_target) ) email_obj_addresses = [] if demo_info.email_objects: for email in demo_info.email_objects: root.append( _build_email_node( email.email_address, is_access_target=email.is_sac_target ) ) email_obj_addresses.append(email.email_address) for email in demo_info.emails: if email in email_obj_addresses: continue root.append( _build_email_node(email, is_access_target=demo_info.email_is_sac_target) ) return etree.tostring(root).decode("utf8")
[docs] def build_add_user_logon_xml( user_id: int, logon_name: str, password: str, skiptac: bool = False, ui_source: str = "OnlineBanking", ) -> str: """ Builds an xml_payload for use with HQ's AddUserLogon :param user_id: Which user this logon belongs to :param logon_name: The string user will login to the online instance with :param password: Password to register with the logon_name :param skiptac: bool to skip tac flow on first login :param ui_source: string that describes the ui source associated with the login :return: Already escaped xml_payload """ root = etree.Element("DalUserLogonEdit") q2_user_logon_node = _build_user_logon_node( user_id, logon_name, password, skiptac=skiptac, ui_source=ui_source ) root.append(q2_user_logon_node) return etree.tostring(root).decode("utf8")
[docs] def build_get_group_id_xml(group_name: str) -> str: """ Builds an xml_payload for use with HQ's GetGroupID :param group_name: Which group to find :return: Already escaped xml_payload """ root = etree.Element("root") group_node = etree.SubElement(root, "Q2_Group") etree.SubElement(group_node, "GroupName").text = group_name return etree.tostring(root).decode("utf8")
[docs] def build_add_recipient_xml( display_name: str, customer_id: int, email_address: str, always_send_email: bool, is_ccd: bool, ach_name: str, is_international=False, ): """ Builds an xml_payload for use with HQ's AddRecipient :param display_name: Friendly name for the recipient :param customer_id: Q2_Customer.CustomerID :param email_address: Q2_Email.EmailAddress :param always_send_email: :param is_ccd: If False, will be set to PPD :param ach_name: Where the money is being sent :param is_international: """ data = etree.Element("Data") recipient = etree.Element("Q2_Recipient") etree.SubElement(recipient, "DisplayName").text = str(display_name) etree.SubElement(recipient, "CustomerID").text = str(customer_id) etree.SubElement(recipient, "EmailAddress").text = str(email_address) etree.SubElement(recipient, "AlwaysSendEmail").text = str(always_send_email).lower() etree.SubElement(recipient, "IsInternational").text = str(is_international).lower() ach_code_text = "PPD" if is_ccd: ach_code_text = "CCD" etree.SubElement(recipient, "ACHClassCode").text = ach_code_text etree.SubElement(recipient, "AchName").text = ach_name etree.SubElement(recipient, "IdentificationNumber").text = None data.append(recipient) return etree.tostring(data).decode("utf8")
[docs] def build_add_recipient_account_xml( recipient_id: int, account_number: int, account_type: str, aba: str, customer_id: int, ): """ Builds an xml_payload for use with HQ's AddRecipientAccount :param recipient_id: Q2_Recipient.RecipientID :param account_number: :param account_type: :param aba: :param customer_id: Q2_Customer.CustomerID """ data = etree.Element("Data") recipient_account = etree.Element("Q2_RecipientAccount") etree.SubElement(recipient_account, "RecipientID").text = str(recipient_id) etree.SubElement(recipient_account, "AccountNumber").text = str(account_number) etree.SubElement(recipient_account, "AccountType").text = str(account_type) etree.SubElement(recipient_account, "ABA").text = str(aba) etree.SubElement(recipient_account, "CustomerID").text = str(customer_id) data.append(recipient_account) return etree.tostring(data).decode("utf8")
[docs] def build_secure_message_xml( source_id: int, target_id: int, message_subject: str, message_body: str, sender_type: Optional[SecureMessengerType] = SecureMessengerType.Administrator, target_type: Optional[SecureMessengerType] = SecureMessengerType.AdministratorGroup, creation_time: Optional[datetime] = None, expiration_time: Optional[datetime] = None, attachment: Optional[SecureMessageAttachment] = None, ): """ Builds an xml_payload for use with HQ's SendSecureMessageAsXML :param source_id: "from" group id :param target_id: recipient ID :param message_subject: Subject of the secure message :param message_body: Body of the secure message :param sender_type: Optional choice from `.SecureMessengerType`. Defaults to Administrator :param target_type: Optional choice from `.SecureMessengerType`. Defaults to AdministratorGroup :param creation_time: Time message was generated :param expiration_time: Timestamp at which message will expire :param attachment: Optional choice from `.SecureMessageAttachment`. :return: Already escaped xml_payload """ root_node = etree.Element( "SecureMessage", xmlns="http://tempuri.org/SecureMessage.xsd" ) secure_message_node = etree.SubElement(root_node, "Message") # From nodes etree.SubElement(secure_message_node, "FromType").text = sender_type etree.SubElement(secure_message_node, "FromID").text = str(source_id) # To nodes etree.SubElement(secure_message_node, "ToType").text = target_type etree.SubElement(secure_message_node, "ToID").text = str(target_id) # Content nodes etree.SubElement(secure_message_node, "Subject").text = message_subject etree.SubElement(secure_message_node, "Body").text = message_body time_format = "%Y-%m-%dT%H:%M:%S" if creation_time is not None: etree.SubElement( secure_message_node, "CreateDate" ).text = creation_time.strftime(time_format) else: # pragma: no cover creation_time = datetime.now() # set a default expiration time of one year if expiration_time is None: expiration_time = creation_time + timedelta(days=365) etree.SubElement( secure_message_node, "ExpirationDate" ).text = expiration_time.strftime(time_format) if attachment: etree.SubElement(secure_message_node, "AttachmentName").text = attachment.name encoded_content = base64.b64encode( base64.b64encode(attachment.body) ) # Yes, it must be double-encoded etree.SubElement(secure_message_node, "AttachmentData").text = encoded_content return etree.tostring(root_node).decode("utf8")
def _build_customer_node( demo_info: DemographicInfo, group_id: int, is_company: bool, host_user: Optional[str] = None, host_pwd: Optional[str] = None, ) -> etree.Element: auto_generated = True auto_generated = str(auto_generated).lower() elem = etree.Element("Q2_Customer") filtered_name = filter( None, [demo_info.first_name, demo_info.middle_name, demo_info.last_name] ) etree.SubElement(elem, "CustomerName").text = " ".join(filtered_name) if demo_info.social_security_number: etree.SubElement(elem, "TaxID").text = demo_info.social_security_number etree.SubElement(elem, "IsCompany").text = str(is_company).lower() etree.SubElement(elem, "AutoGenerated").text = auto_generated etree.SubElement(elem, "GroupID").text = str(group_id) etree.SubElement(elem, "CustInfo").text = demo_info.social_security_number etree.SubElement(elem, "PrimaryCIF").text = ( demo_info.primary_cif if demo_info.primary_cif else demo_info.social_security_number ) if host_user: etree.SubElement(elem, "HostUser").text = host_user if host_pwd: etree.SubElement(elem, "HostPwd").text = host_pwd return elem def _build_user_node( demo_info: DemographicInfo, customer_id: int, host_user: Optional[str] = None, host_pwd: Optional[str] = None, null_primary_cif: bool = False, ) -> etree.Element: auto_generated = True auto_generated = str(auto_generated).lower() elem = etree.Element("Q2_User") etree.SubElement(elem, "CustomerID").text = str(customer_id) etree.SubElement(elem, "FirstName").text = demo_info.first_name etree.SubElement(elem, "MiddleName").text = demo_info.middle_name etree.SubElement(elem, "LastName").text = demo_info.last_name if demo_info.social_security_number: etree.SubElement(elem, "SSN").text = demo_info.social_security_number etree.SubElement(elem, "AutoGenerated").text = auto_generated user_info = demo_info.user_info or demo_info.social_security_number if user_info: try: user_info = str(user_info) etree.SubElement(elem, "UserInfo").text = user_info except ValueError: pass primary_cif = ( demo_info.primary_cif if demo_info.primary_cif else demo_info.social_security_number ) if null_primary_cif: primary_cif = None if primary_cif: etree.SubElement(elem, "PrimaryCIF").text = ( demo_info.primary_cif if demo_info.primary_cif else demo_info.social_security_number ) etree.SubElement(elem, "IsAdmin").text = "1" if demo_info.is_admin else "0" if demo_info.dob: etree.SubElement(elem, "DOB").text = demo_info.dob if demo_info.user_role_id: etree.SubElement(elem, "UserRoleID").text = str(demo_info.user_role_id) if demo_info.auth_token_serial: etree.SubElement(elem, "AuthTokenSerial").text = demo_info.auth_token_serial if host_user: etree.SubElement(elem, "HostUser").text = host_user if host_pwd: etree.SubElement(elem, "HostPwd").text = host_pwd return elem def _build_user_logon_node( user_id: int, logon_name: str, password: str, skiptac: bool = False, ui_source: str = "OnlineBanking", ) -> etree.Element: auto_generated = True auto_generated = str(auto_generated).lower() elem = etree.Element("Q2_UserLogon") etree.SubElement(elem, "UserID").text = str(user_id) etree.SubElement(elem, "LoginName").text = logon_name etree.SubElement(elem, "UserPassword").text = password etree.SubElement(elem, "UISource").text = ui_source etree.SubElement(elem, "AutoGenerated").text = auto_generated etree.SubElement(elem, "SkipTacFirstLogin").text = str(skiptac).lower() return elem def _build_address_node(address: Address) -> etree.Element: elem = etree.Element("Q2_Address") etree.SubElement(elem, "StreetAddress1").text = address.address_1 etree.SubElement(elem, "StreetAddress2").text = address.address_2 etree.SubElement(elem, "City").text = address.city if address.country.upper() != "USA": etree.SubElement(elem, "IsInternational").text = "true" state = address.state province = address.province if province: etree.SubElement(elem, "Province").text = province elif state: etree.SubElement(elem, "Province").text = state else: etree.SubElement(elem, "State").text = address.state etree.SubElement(elem, "PostalCode").text = address.zipcode etree.SubElement(elem, "CountryCode").text = address.country etree.SubElement(elem, "AddressType").text = address.address_type return elem def _build_phone_node(phone: Phone, with_access=False, with_sms=False) -> etree.Element: is_access_target = "true" is_sms_target = "false" if phone.type == PhoneType.CELL: is_sms_target = "true" elem = etree.Element("Q2_PhoneNumber") etree.SubElement(elem, "CountryCode").text = phone.country etree.SubElement(elem, "CityOrAreaCode").text = phone.area_code etree.SubElement(elem, "LocalNumber").text = phone.phone_number ext_node = etree.SubElement(elem, "Extension") if phone.extension: ext_node.text = phone.extension etree.SubElement(elem, "PhoneType").text = phone.type if with_access: etree.SubElement(elem, "IsAccessTarget").text = is_access_target if with_sms: etree.SubElement(elem, "IsSmsTarget").text = is_sms_target return elem def _build_email_node( email: str, is_access_target: Optional[bool] = None ) -> etree.Element: is_access_target = is_access_target is None or is_access_target assert isinstance(is_access_target, bool), "is_access_target must be a boolean" elem = etree.Element("Q2_Email") etree.SubElement(elem, "EmailAddress").text = email etree.SubElement(elem, "IsAccessTarget").text = str(is_access_target).lower() return elem def _build_subsidiary_node(subsidiary: Subsidiary) -> etree.Element: elem = etree.Element("Q2_Subsidiary") etree.SubElement(elem, "DisplayName").text = subsidiary.display_name if subsidiary.ach_info: etree.SubElement(elem, "AchName").text = subsidiary.ach_info.name etree.SubElement(elem, "AchTaxId").text = subsidiary.ach_info.tax_id if subsidiary.wire_info: etree.SubElement(elem, "WireName").text = subsidiary.wire_info.name etree.SubElement( elem, "WireAddress1" ).text = subsidiary.wire_info.address.address_1 etree.SubElement( elem, "WireAddress2" ).text = subsidiary.wire_info.address.address_2 etree.SubElement(elem, "WireCity").text = subsidiary.wire_info.address.city country_code = subsidiary.wire_info.address.country if country_code.upper() != "USA": province = subsidiary.wire_info.address.province if not province: province = subsidiary.wire_info.address.state etree.SubElement(elem, "WireProvince").text = province else: etree.SubElement( elem, "WireState" ).text = subsidiary.wire_info.address.state etree.SubElement( elem, "WirePostalCode" ).text = subsidiary.wire_info.address.zipcode etree.SubElement(elem, "WireCountryCode").text = country_code etree.SubElement(elem, "WireIsInternational").text = ( "1" if subsidiary.wire_info.wire_is_international else "0" ) return elem def build_account_association_xml( customer_id: int, user_id: int, account_access: int, host_account_id: Optional[int] = None, link_by_cif=False, cif_internal: Optional[str] = None, customer_account_id: Optional[int] = None, primary_only: Optional[bool] = True, user_role_id: Optional[int] = None, ) -> str: customer_id = str(customer_id) user_id = str(user_id) account_access = str(account_access) add_account_request = etree.Element("AccountAssociation") user_account = etree.SubElement(add_account_request, "Q2_UserAccount") user_account_node = etree.SubElement(user_account, "UserID") user_account_node.text = user_id user_access_node = etree.SubElement(user_account, "Access") user_access_node.text = account_access if user_role_id: role_id = str(user_role_id) user_role_node = etree.SubElement(user_account, "UserRoleID") user_role_node.text = role_id if not link_by_cif: haid = str(host_account_id) account_node = etree.SubElement(user_account, "HostAccountID") account_node.text = haid if not customer_account_id: customer_account = etree.SubElement(add_account_request, "Q2_CustomerAccount") customer_id_node = etree.SubElement(customer_account, "CustomerID") customer_id_node.text = customer_id access_node = etree.SubElement(customer_account, "Access") access_node.text = account_access link_by_cif_node = etree.SubElement(customer_account, "LinkByCIF") link_by_cif_node.text = str(link_by_cif).lower() if not link_by_cif: haid = str(host_account_id) account_node = etree.SubElement(customer_account, "HostAccountID") account_node.text = haid else: cif_internal_node = etree.SubElement(customer_account, "CifInternal") cif_internal_node.text = cif_internal primary_only_node = etree.SubElement(customer_account, "PrimaryOnly") primary_only_node.text = str(primary_only) else: customer_account_node = etree.SubElement(user_account, "CustomerAccountID") customer_account_node.text = str(customer_account_id) return etree.tostring(add_account_request).decode()
[docs] def build_update_demographics_by_logon_name_xml( login_name: str, demographic_info: DemographicInfo ) -> str: """ Transforms login_name and DemographicInfo into an xml_payload for use with apispUpdateDemographicsByLogonName_RT stored procedure :param login_name: The string user will login to the online instance with :param demographic_info: Instance of q2_sdk.models.demographic.DemographicInfo """ demo_inf = demographic_info demographic_node = etree.Element("Demographics") update_user_flag = ( "1" if any([demo_inf.first_name, demo_inf.middle_name, demo_inf.last_name]) else "0" ) user_node = etree.SubElement( demographic_node, "User", loginName=login_name, FirstName=demo_inf.first_name, MiddleName=demo_inf.middle_name, LastName=demo_inf.last_name, updateUserRow=update_user_flag, ) if demo_inf.dob: user_node.set("DOB", demo_inf.dob) if demo_inf.emails: etree.SubElement(user_node, "Email", address=demo_inf.emails[0]) if demo_inf.addresses: first_address = demo_inf.addresses[0] etree.SubElement( user_node, "Address", country=first_address.country, street1=first_address.address_1, street2=first_address.address_2, state=first_address.state, city=first_address.city, zip=first_address.zipcode, province=first_address.province, addressType=first_address.address_type, ) for phone in demo_inf.phones: etree.SubElement( user_node, "Phone", loginName=login_name, country=phone.country, area=phone.area_code, number=phone.phone_number, type=phone.type, extension=phone.extension, ) return etree.tostring(demographic_node).decode()
[docs] def generate_user_logon_password(policy_obj: PasswordPolicy) -> str: """ Uses the parameters of the password policy to generate a valid password to be used when creating a login :return: a valid password """ similar = { "i", "I", "1", "l", "0", "o", "O", "|", "'", "`", "{", "}", "(", ")", "[", "]", ".", ",", '"', "\\", ";", ":", } if not policy_obj.excluded_characters: policy_obj.excluded_characters = similar else: for char in policy_obj.excluded_characters: similar.add(char) policy_obj.excluded_characters = similar lower_character = set(string.ascii_lowercase) upper_character = set(string.ascii_uppercase) numbers = set(string.digits) special = set(string.punctuation) lower_character -= policy_obj.excluded_characters upper_character -= policy_obj.excluded_characters numbers -= policy_obj.excluded_characters special -= policy_obj.excluded_characters required_groups = [lower_character] if policy_obj.upper_required: required_groups.append(upper_character) if policy_obj.numbers_required: required_groups.append(numbers) if policy_obj.special_required: required_groups.append(special) password_composition = [] for group in required_groups: next_character = random.choice(list(group)) password_composition += next_character if policy_obj.limit_adjacent or policy_obj.limit_repeating: group.discard(next_character) if policy_obj.min_length > policy_obj.max_length: policy_obj.min_length = policy_obj.max_length all_characters = list(set.union(*required_groups)) password_length = policy_obj.min_length filler_length = min( password_length - len(password_composition), len(all_characters) ) for _ in range(filler_length): next_character = random.choice(all_characters) password_composition += next_character if policy_obj.limit_adjacent or policy_obj.limit_repeating: all_characters.remove(next_character) random.shuffle(password_composition) return "".join(password_composition)
[docs] def get_frequency_bit_flags( days_of_week: Optional[list[DayOfWeek]] = None, weeks_of_month: Optional[list[WeekOfMonth]] = None, days_of_month: Optional[list[int]] = None, ) -> Optional[int]: """ :param days_of_week: List of `q2_sdk.hq.models.transaction_info.DayOfWeek` objects :param weeks_of_month: List of `q2_sdk.hq.models.transaction_info.WeekOfMonth` objects :param days_of_month: ex. [1, 15] for 1st and 15th of month For use with the various AddRecurring* modules in HqApi Example usage: .. testcode:: from q2_sdk.hq.models.transaction_info import DayOfWeek, WeekOfMonth flags = get_frequency_bit_flags( [ DayOfWeek.Monday, DayOfWeek.Tuesday ], [ WeekOfMonth.First ] ) This would set ``flags`` to an integer that meant 'Monday and Tuesday on the First week of the month'. """ by_week_of_month = all([ days_of_week is not None, weeks_of_month is not None, days_of_month is None, ]) by_day_of_month = all([ days_of_week is None, weeks_of_month is None, days_of_month is not None, ]) error_msg = "Must provide either (days_of_week and weeks_of_month) or days_of_month" assert any([by_week_of_month, by_day_of_month]), error_msg if by_week_of_month: days_int = sum([x.value for x in days_of_week]) weeks_int = sum([x.value for x in weeks_of_month]) return days_int + weeks_int elif by_day_of_month: bitarr = ["0"] * 31 for i in days_of_month: bitarr[i - 1] = "1" bitstr = "".join(reversed(bitarr)) return int( bitstr, 2 ) # Treat the bitstr as a base 2 number, then transform it to decimal
def build_policy_data_from_hq_response(result_node) -> PolicyData: policy_data = PolicyData() for node in result_node.Data.PolicyData.getchildren(): object_name = node.tag.lstrip("Q2_Policy") values_dict = {x.tag: x.pyval for x in node.getchildren()} level, identifier = values_dict["PolicyIdentifier"].split("-") match level: # noqa: E999 case "U": if not policy_data.User: policy_data.User = User(ID=int(identifier)) add_policy_data_to_entity(object_name, values_dict, policy_data.User) case "C": if not policy_data.Customer: policy_data.Customer = Customer(ID=int(identifier)) add_policy_data_to_entity( object_name, values_dict, policy_data.Customer ) case "G": if not policy_data.Group: policy_data.Group = Group(ID=int(identifier)) add_policy_data_to_entity(object_name, values_dict, policy_data.Group) case "R": if not policy_data.UserRole: policy_data.UserRole = UserRole(ID=int(identifier)) add_policy_data_to_entity( object_name, values_dict, policy_data.UserRole ) case "O": if not policy_data.Company: policy_data.Company = Company(ID=int(identifier)) add_policy_data_to_entity(object_name, values_dict, policy_data.Company) return policy_data def add_policy_data_to_entity(object_name, values_dict, entity): match object_name: case "GeneratedTransactionRights": data = GeneratedTransactionRights.from_kwargs(**values_dict) short_name = data.ShortName if not entity.GeneratedTransactionRights: entity.GeneratedTransactionRights = [] entity.GeneratedTransactionRights.append(data) if not entity.TransactionRightsShortNameMapping: entity.TransactionRightsShortNameMapping = {} entity.TransactionRightsShortNameMapping[short_name] = data case "Subsidiaries": data = Subsidiaries.from_kwargs(**values_dict) if not entity.Subsidiaries: entity.Subsidiaries = [] entity.Subsidiaries.append(data) case "Accounts": data = Account.from_kwargs(**values_dict) host_account_id = data.HostAccountID if not entity.Accounts: entity.Accounts = [] entity.Accounts.append(data) if not entity.AccountIdMapping: entity.AccountIdMapping = {} entity.AccountIdMapping[host_account_id] = data case "Features": data = Features.from_kwargs(**values_dict) short_name = data.PropertyName if not entity.Features: entity.Features = [] entity.Features.append(data) if not entity.FeaturesShortNameMapping: entity.FeaturesShortNameMapping = {} entity.FeaturesShortNameMapping[short_name] = data case "Data": data = Data.from_kwargs(**values_dict) if not entity.Data: entity.Data = [] entity.Data.append(data)