"""
Realtime Payments, or RTP are handled both as an adapter through the Online Banking
interface, as well as through inbound calls from an outside institution.
If a end user at a Q2 institution is kicking off the flow through their Online
Banking, then it uses the adapter flow. However, they are sending the money
somewhere, perhaps another institution. That other institution needs to accept
the request and do something on its end. That's the other flow, handling the
inbound calls.
In both cases, we use the Request objects in here enumerated in the RequestType
class in this file. Insantiating these objects can be done with .from_hq_request
or .from_external_request. Both take in a dictionary (or json blob) and hydrate
into the object. For inbound calls, these have a .store_in_hq method that will fill
the Q2_RealtimePaymentExternal table in the database, which is viewable in the
Online Banking interface as well.
"""
from __future__ import annotations
import json
from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
from enum import Enum
from dateutil import parser
from typing import Optional
from q2_sdk.core import contexts
from q2_sdk.core.exceptions import SystemDependencyError
from q2_sdk.hq.db.instant_payments import InstantPayments
from q2_sdk.models.demographic import Address, Address3
from q2_sdk.hq.hq_api.q2_api import AcceptIncomingRealTimePaymentTransaction
from q2_sdk.hq.models.hq_response import HqResponse
Q2MSG_SUPPORT = False
try:
from google.protobuf.message import Message as ProtobufMessage
from q2msg.services.realtimepayments_pb2 import (
RealtimeResponse,
PaymentIdentification,
Error,
)
Q2MSG_SUPPORT = True
except ImportError: # pragma: no cover
RealtimeResponse = None
PaymentIdentification = None
Error = None
[docs]
@dataclass
class Agent:
routing_number: str = None
[docs]
@dataclass
class Q2Entity:
phone_number: str
organization: dict
agent: Agent = None
account_number: str = None
name: str = None
address: Address = None
email: str = None
country_of_residence: str = None
[docs]
@dataclass
class OtherParty:
phone_number: str
organization: dict
agent: Agent = None
account_number: str = None
name: str = None
address: Address = None
email: str = None
country_of_residence: str = None
[docs]
@dataclass
class InstantPaymentsIdentification:
end_to_end_identification: str = None
instruction_identification: str = None
transaction_identification: str = None
[docs]
@dataclass
class CommonRequestData:
generated_transaction_id: int
user_id: int
customer_id: int
debtor_address: Address
creditor_address: Address
[docs]
@staticmethod
def _from_hq_request(inp: dict) -> CommonRequestData:
generated_transaction_id = inp["generatedTransactionId"]
user_id = inp["userId"]
customer_id = inp["customerId"]
debtor_address = CommonRequestData._get_address_obj(inp, "debtor")
creditor_address = CommonRequestData._get_address_obj(inp, "creditor")
return CommonRequestData(
generated_transaction_id,
user_id,
customer_id,
debtor_address,
creditor_address,
)
[docs]
@staticmethod
def _get_address_obj(inp: dict, address_loc: str) -> Address:
address_node = inp[address_loc].get("address", {})
address_1 = address_node.get("address1", "")
address_2 = address_node.get("address2", "")
(city, zipcode, state, country) = ("", "", "", "")
if "address3" in address_node:
address_3 = Address3.from_str(address_node["address3"])
city = address_3.city
zipcode = address_3.zipcode
state = address_3.state
country = address_3.country
return Address(address_1, address_2, city, state, zipcode, country=country)
[docs]
@staticmethod
def _get_payment_identification(inp: dict) -> InstantPaymentsIdentification:
payment_identification = inp.get("paymentIdentification", {})
# assert payment_identification.get(
# "endToEndIdentification"
# ) or payment_identification.get("instructionIdentification"), (
# "Either endToEndIdentification or instructionIdentification must be provided"
# )
return InstantPaymentsIdentification(
payment_identification.get("endToEndIdentification"),
payment_identification.get("instructionIdentification"),
payment_identification.get("transactionIdentification"),
)
[docs]
@dataclass
class CreditTransferRequest:
generated_transaction_id: int
user_id: int
customer_id: int
debtor: OtherParty
amount: Decimal
currency_code: str
creditor: Q2Entity
settlement_date: Optional[datetime]
payment_identification: InstantPaymentsIdentification = None
remittance_information: dict = None
[docs]
@staticmethod
def from_external_request(inp: dict) -> CreditTransferRequest:
inp["generatedTransactionId"] = -1
inp["userId"] = -1
inp["customerId"] = -1
return CreditTransferRequest.from_hq_request(inp)
[docs]
@staticmethod
def from_hq_request(inp: dict) -> CreditTransferRequest:
common_data = CommonRequestData._from_hq_request(inp)
contact_details = inp["debtor"].get("contactDetails", {})
other_email = contact_details.get("emailAddress")
other_acct = inp.get("debtorAccountNumber", "")
debtor_agent = inp.get("debtorAgent", {})
creditor_agent = inp.get("creditorAgent", {})
debtor = OtherParty(
inp["debtor"].get("contactDetails", {}).get("phoneNumber"),
inp["debtor"].get("organizationIdentification", {}),
Agent(debtor_agent.get("financialInstitution", {}).get("routingNumber")),
other_acct,
inp["debtor"]["name"],
common_data.debtor_address,
other_email,
inp["creditor"].get("countryOfResidence"),
)
creditor = Q2Entity(
inp["creditor"].get("contactDetails", {}).get("phoneNumber"),
inp["creditor"].get("organizationIdentification", {}),
Agent(creditor_agent.get("financialInstitution", {}).get("routingNumber")),
inp.get("creditorAccountNumber"),
inp["creditor"]["name"],
common_data.creditor_address,
inp["creditor"].get("contactDetails", {}).get("emailAddress"),
inp["creditor"].get("countryOfResidence"),
)
payment_identification = CommonRequestData._get_payment_identification(inp)
amount = Decimal(str(inp["amount"]))
currency_code = inp["currencyCode"]
settlement_date = inp.get("settlementDate")
if settlement_date:
settlement_date = parser.parse(settlement_date)
return CreditTransferRequest(
common_data.generated_transaction_id,
common_data.user_id,
common_data.customer_id,
debtor,
amount,
currency_code,
creditor,
settlement_date,
payment_identification,
inp.get("remittanceInformation"),
)
[docs]
async def store_in_hq(self, vendor_name: str) -> HqResponse:
"""
:param: vendor_name: Matches Q2_RTPVendor.ShortName
"""
unique_identifier = _get_unique_identifier(self.payment_identification)
logger = contexts.get_current_request().request_handler.logger
settlement_date = None
remittance_information_str = None
rtp_vendor = await InstantPayments(logger).get_vendor_by_name(vendor_name)
vendor_identifier = str(rtp_vendor.StringIdentifier)
if self.remittance_information:
remittance_information_str = json.dumps(self.remittance_information)
if self.settlement_date:
settlement_date = self.settlement_date.strftime("%m/%d/%Y")
debtor_country = (
self.debtor.country_of_residence
if self.debtor.country_of_residence
else self.debtor.address.country
)
params_obj = AcceptIncomingRealTimePaymentTransaction.ParamsObj(
logger,
float(self.amount),
False,
vendor_name=vendor_identifier,
request_type_short_name="CreditTransfer",
currency_code=self.currency_code,
remote_entity_name=self.debtor.name,
remote_entity_address1=self.debtor.address.address_1,
remote_entity_address2=self.debtor.address.address_2,
remote_entity_city=self.debtor.address.city,
remote_entity_postal_code=self.debtor.address.zipcode,
remote_entity_state=self.debtor.address.state,
remote_entity_country_code=debtor_country,
remote_entity_email_address=self.debtor.email,
remote_fi_account_number=self.debtor.account_number,
local_entity_name=self.creditor.name,
local_external_account_number=self.creditor.account_number,
requested_execution_date=settlement_date,
service_call_conversation_identifier=unique_identifier,
description=remittance_information_str,
)
return await AcceptIncomingRealTimePaymentTransaction.execute(params_obj)
[docs]
@dataclass
class RequestForPaymentRequest:
generated_transaction_id: int
user_id: int
customer_id: int
debtor: Q2Entity
amount: Decimal
currency: str
condition: dict
creditor: OtherParty
expiry: Optional[datetime]
payment_identification: InstantPaymentsIdentification
remittance_information: dict = None
[docs]
@staticmethod
def from_external_request(inp: dict) -> RequestForPaymentRequest:
inp["generatedTransactionId"] = -1
inp["userId"] = -1
inp["customerId"] = -1
return RequestForPaymentRequest.from_hq_request(inp)
[docs]
@staticmethod
def from_hq_request(inp: dict) -> RequestForPaymentRequest:
common_data = CommonRequestData._from_hq_request(inp)
contact_details = inp["debtor"].get("contactDetails", {})
other_email = contact_details.get("emailAddress")
other_acct = inp.get("debtorAccountNumber", "")
debtor_agent = inp.get("debtorAgent", {})
creditor_agent = inp.get("creditorAgent", {})
debtor = Q2Entity(
inp["debtor"].get("contactDetails", {}).get("phoneNumber"),
inp["debtor"].get("organizationIdentification", {}),
Agent(debtor_agent.get("financialInstitution", {}).get("routingNumber")),
other_acct,
inp["debtor"]["name"],
common_data.debtor_address,
other_email,
)
amount = Decimal(str(inp["instructedAmount"]))
currency = inp["instructedCurrency"]
condition = inp["paymentCondition"]
contact_details = inp["creditor"].get("contactDetails", {})
other_email = contact_details.get("emailAddress")
other_acct = inp.get("creditorAccountNumber", "")
creditor = OtherParty(
inp["creditor"].get("contactDetails", {}).get("phoneNumber"),
inp["creditor"].get("organizationIdentification", {}),
Agent(creditor_agent.get("financialInstitution", {}).get("routingNumber")),
other_acct,
inp["creditor"].get("name"),
common_data.creditor_address,
other_email,
)
payment_identification = CommonRequestData._get_payment_identification(inp)
expiry = inp.get("expiryDate")
if expiry:
expiry = parser.parse(expiry)
return RequestForPaymentRequest(
common_data.generated_transaction_id,
common_data.user_id,
common_data.customer_id,
debtor,
amount,
currency,
condition,
creditor,
expiry,
payment_identification,
inp.get("remittanceInformation"),
)
[docs]
async def store_in_hq(self, vendor_name: str) -> HqResponse:
"""
:param vendor_name: Matches Q2_RTPVendor.ShortName
"""
unique_identifier = _get_unique_identifier(self.payment_identification)
logger = contexts.get_current_request().request_handler.logger
rtp_vendor = await InstantPayments(logger).get_vendor_by_name(vendor_name)
vendor_identifier = str(rtp_vendor.StringIdentifier)
expiry = None
remittance_information_str = None
if self.remittance_information:
remittance_information_str = json.dumps(self.remittance_information)
if self.expiry:
expiry = self.expiry.isoformat()
params_obj = AcceptIncomingRealTimePaymentTransaction.ParamsObj(
logger,
float(self.amount),
False,
vendor_name=vendor_identifier,
request_type_short_name="RequestForPayment",
currency_code=self.currency,
remote_entity_name=self.creditor.name,
remote_entity_address1=self.creditor.address.address_1,
remote_entity_address2=self.creditor.address.address_2,
remote_entity_city=self.creditor.address.city,
remote_entity_postal_code=self.creditor.address.zipcode,
remote_entity_state=self.creditor.address.state,
remote_entity_country_code=self.creditor.address.country,
remote_entity_email_address=self.creditor.email,
remote_fi_account_number=self.creditor.account_number,
local_entity_name=self.debtor.name,
local_external_account_number=self.debtor.account_number,
expiry_date=expiry,
service_call_conversation_identifier=unique_identifier,
description=remittance_information_str,
)
return await AcceptIncomingRealTimePaymentTransaction.execute(params_obj)
[docs]
class RequestType(Enum):
CreditTransferRequest = CreditTransferRequest
RealtimePaymentRequest = RequestForPaymentRequest
[docs]
@dataclass
class InstantPaymentsResponse:
success: bool
external_id: str
exception_message: str = ""
[docs]
def serialize_as_q2msg(self) -> ProtobufMessage:
if Q2MSG_SUPPORT is None: # pragma: no cover
raise SystemDependencyError(
"q2msg version >= 1.21.4 must be installed to use InstantPayments"
)
match self.success:
case True:
error_return_code = 0
case False:
error_return_code = -1
errors = []
if not self.success:
errors = [
Error(code=str(error_return_code), message=self.exception_message)
]
return RealtimeResponse(
success=self.success,
payment_identification=PaymentIdentification(
end_to_end_identification=self.external_id
),
error_return_code=error_return_code,
end_user_message="",
errors=errors,
)
[docs]
def serialize_as_dict(self):
match self.success:
case True:
error_return_code = 0
case False:
error_return_code = -1
errors = []
if not self.success:
errors.append({
"code": str(error_return_code),
"message": self.exception_message,
})
return {
"success": self.success,
"data": {
"payment_identification": {
"end_to_end_identification": self.external_id
},
"errors": errors,
},
"errorReturnCode": error_return_code,
"endUserMessage": "",
"expectedWaitTime": 0,
}
[docs]
def _get_unique_identifier(payment_identification: InstantPaymentsIdentification):
if not payment_identification.end_to_end_identification:
return payment_identification.instruction_identification
return payment_identification.end_to_end_identification
# Aliases to keep client code tight and allow for expanding in the future
CreditTransferResponse = InstantPaymentsResponse
RequestForPaymentResponse = InstantPaymentsResponse