Source code for q2_cores.JXchange.core

import logging
from typing import List, Optional

from q2_sdk.hq.models.hq_credentials import HqCredentials
from q2_sdk.models.cores.base_core import BaseCore
from q2_sdk.models.cores.models.core_user import CoreUser
from q2_sdk.models.demographic import DemographicInfo
from q2_cores.JXchange import mappers, queries




[docs]
class Core(BaseCore):
    CONFIG_FILE_NAME = "JXchange_Core"
    OPTIONAL_CONFIGURATIONS = {
        "CHANGE_ELEMENT_TO_MsgRqHdr": False,
    }

    def __init__(
        self,
        logger: logging.Logger,
        core_user: CoreUser,
        hq_credentials: Optional[HqCredentials] = None,
        **kwargs,
    ):
        super().__init__(logger, core_user, hq_credentials=hq_credentials, **kwargs)



[docs]
    async def build_demographic_info(self) -> mappers.DemographicInfoMapper:
        change_tag = getattr(self.config, "CHANGE_ELEMENT_TO_MsgRqHdr", False)
        query = queries.DemographicInfoQuery(
            self.logger, self.core_user.online_user.customer_primary_cif, change_tag
        )
        mapper = mappers.DemographicInfoMapper(
            [query], hq_credentials=self.hq_credentials, zone_context=self.core_user
        )
        return mapper





[docs]
    async def build_update_demographic_info(
        self, demographic_info: DemographicInfo
    ) -> mappers.UpdateDemographicInfoMapper:
        query = queries.UpdateDemographicInfoQuery(
            self.logger,
            self.core_user.online_user.customer_primary_cif,
            demographic_info,
        )
        mapper = mappers.UpdateDemographicInfoMapper(
            [query], hq_credentials=self.hq_credentials, zone_context=self.core_user
        )
        return mapper








[docs]
    async def build_account_inquiry(
        self,
        acct_id: str,
        acct_type: str,
        xtend_elems: Optional[List[str]] = None,
    ) -> mappers.AccountInquiryMapper:
        """
        Retrieve account details including address information from xtend
        elements. Address fields for deposit accounts are under
        ``x_DepInfoRec.Addr``; pass a custom ``xtend_elems`` list to adjust
        the payload.
        """
        query = queries.AccountInquiryQuery(
            self.logger,
            acct_id=acct_id,
            acct_type=acct_type,
            xtend_elems=xtend_elems,
        )
        return mappers.AccountInquiryMapper(
            [query], hq_credentials=self.hq_credentials, zone_context=self.core_user
        )





[docs]
    async def build_address_modify(
        self,
        addr_key: str,
        new_address: Optional[dict] = None,
        delete_address: bool = False,
    ) -> mappers.AddressModifyMapper:
        """
        Modify or delete an existing address record. ``addr_key`` is always
        required. Pass ``new_address`` (dict with ``address_1``, ``city``,
        ``state``, ``zip_code``, and optional ``address_2``) to update, or
        set ``delete_address=True`` to remove the record.
        """
        query = queries.AddressModifyQuery(
            self.logger,
            addr_key=addr_key,
            new_address=new_address,
            delete_address=delete_address,
        )
        return mappers.AddressModifyMapper(
            [query], hq_credentials=self.hq_credentials, zone_context=self.core_user
        )





[docs]
    async def build_address_add(
        self,
        new_address: dict,
        addr_cat_use: str,
        cust_id: Optional[str] = None,
        acct_id: Optional[str] = None,
        acct_type: Optional[str] = None,
        addr_cat_2_use: Optional[str] = None,
    ) -> mappers.AddressAddMapper:
        """
        Add a new alternate address record. Provide ``cust_id`` for
        CIF-level addresses or ``acct_id`` + ``acct_type`` for account-level.
        ``addr_cat_use`` is required (e.g. ``'Cust'``, ``'Acct'``,
        ``'Seasonal'``). The mapper returns the new ``AddrKey``.
        """
        query = queries.AddressAddQuery(
            self.logger,
            new_address=new_address,
            addr_cat_use=addr_cat_use,
            cust_id=cust_id,
            acct_id=acct_id,
            acct_type=acct_type,
            addr_cat_2_use=addr_cat_2_use,
        )
        return mappers.AddressAddMapper(
            [query], hq_credentials=self.hq_credentials, zone_context=self.core_user
        )