from dataclasses import dataclass
from typing import Optional
from q2_sdk.core import contexts
from q2_sdk.hq.db.instant_payments import InstantPayments
from q2_sdk.hq.hq_api.q2_api import AcceptIncomingRealTimePaymentTransaction
from q2_sdk.hq.models.hq_response import HqResponse
[docs]
@dataclass
class PostalAddress:
address1: str
address2: str
address3: str
[docs]
@dataclass
class DateAndPlaceOfBirth:
birth_date: str
province_of_birth: str
city_of_birth: str
[docs]
@dataclass
class OrganizationIdentification:
name: str
any_bic: str
legal_entity_identification: str
[docs]
@dataclass
class PartyIdentification:
name: str
address: PostalAddress
date_and_place_of_birth: DateAndPlaceOfBirth
organization_identification: OrganizationIdentification
country_of_residence: str
city_of_residence: str
contact_details: ContactDetails
[docs]
@dataclass
class FinancialInstitutionIdentification:
bicFi: str
routing_number: str
name: str
address: PostalAddress
[docs]
@dataclass
class BranchIdentification:
name: str
address: PostalAddress
[docs]
@dataclass
class BranchAndFinancialInstitutionIdentification:
financial_institution: FinancialInstitutionIdentification
branch: BranchIdentification
[docs]
@dataclass
class CodeOrPropietaryType:
code: str
propietary: str
issuer: str
[docs]
@dataclass
class OriginalGroupInfoAndStatus:
messageIdentification: str
messageNameIdentification: str
creationDatetime: str
numberOfTransactions: int
controlSum: float
statusReasonInformation: list[StatusReasonInformation]
[docs]
@dataclass
class PaymentIdentification:
end_to_end_identification: str
transaction_identification: Optional[str] = None
instruction_identification: Optional[str] = None
[docs]
@dataclass
class PaymentCondition:
amount_modification_allowed: bool
early_payment_allowed: bool
delay_penalty: str
immediate_payment_rebate: ImmediatePaymentRebate
guaranteed_payment_requested: bool
[docs]
@dataclass
class OriginalTransactionReference:
amount: float
expiry_date: str
paymentCondition: PaymentCondition
[docs]
@dataclass
class PaymentConditionStatus:
accepted_amount: float
[docs]
@dataclass
class RequestForPaymentResponse:
initiatingParty: PartyIdentification
debtor_agent: BranchAndFinancialInstitutionIdentification
creditor_agent: BranchAndFinancialInstitutionIdentification
original_group_info_and_status: OriginalGroupInfoAndStatus
original_payment_info_and_status: OriginalPaymentInformationAndStatus
forwarding_agent: Optional[BranchAndFinancialInstitutionIdentification] = None
payment_identification: Optional[PaymentIdentification] = None
[docs]
async def store_in_hq(self, vendor_name: str) -> HqResponse:
"""
This method is used to store the responses for RequestForPayment request in HQ
:param vendor_name: Matches Q2_RTPVendor.ShortName
"""
logger = contexts.get_current_request().request_handler.logger
rtp_vendor = await InstantPayments(logger).get_vendor_by_name(vendor_name)
vendor_identifier = rtp_vendor.StringIdentifier
params_obj = AcceptIncomingRealTimePaymentTransaction.ParamsObj(
logger,
self.original_payment_info_and_status.transaction_info_and_status[
0
].payment_condition_status.accepted_amount,
False,
vendor_name=str(vendor_identifier),
request_type_short_name="RequestForPaymentResponse",
remote_entity_name=self.debtor_agent.branch.name,
remote_entity_address1=self.debtor_agent.branch.address.address1,
remote_entity_address2=self.debtor_agent.branch.address.address2,
remote_entity_email_address=self.initiatingParty.contact_details.email_address,
remote_fi_name=self.debtor_agent.financial_institution.name,
remote_fi_routing_number=self.debtor_agent.financial_institution.routing_number,
remote_entity_identifier=self.debtor_agent.financial_institution.bicFi,
local_entity_name=self.creditor_agent.financial_institution.name,
local_entity_identifier=self.creditor_agent.financial_institution.bicFi,
local_entity_address1=self.creditor_agent.financial_institution.address.address1,
local_entity_address2=self.creditor_agent.financial_institution.address.address2,
local_entity_address3=self.creditor_agent.financial_institution.address.address3,
service_call_unique_identifier=self.payment_identification.end_to_end_identification,
)
return await AcceptIncomingRealTimePaymentTransaction.execute(params_obj)