Reference

shared.services.payment_providers.jpmorgan.entities

JPMorganAccountType

Bases: AlanBaseEnum

IBAN class-attribute instance-attribute

IBAN = 'IBAN'

INTERAC class-attribute instance-attribute

INTERAC = 'INTERAC'

JPMorganBusinessAccountConfig dataclass

JPMorganBusinessAccountConfig(
    base_url,
    account_id,
    debtor_agent_bic,
    signature_private_key,
    signature_private_key_secret_name,
    transport_public_certificate,
    transport_public_certificate_secret_name,
    transport_private_key,
    transport_private_key_secret_name,
)

account_id instance-attribute

account_id

base_url instance-attribute

base_url

debtor_agent_bic instance-attribute

debtor_agent_bic

signature_private_key instance-attribute

signature_private_key

signature_private_key_secret_name instance-attribute

signature_private_key_secret_name

transport_private_key instance-attribute

transport_private_key

transport_private_key_secret_name instance-attribute

transport_private_key_secret_name

transport_public_certificate instance-attribute

transport_public_certificate

transport_public_certificate_secret_name instance-attribute

transport_public_certificate_secret_name

JPMorganBusinessAccountName

Bases: AlanBaseEnum

alan_insurance class-attribute instance-attribute

alan_insurance = 'alan_insurance'

JPMorganCallbackException dataclass

JPMorganCallbackException(
    errorCode, errorDescription, ruleDefinition
)

errorCode instance-attribute

errorCode

errorDescription instance-attribute

errorDescription

ruleDefinition instance-attribute

ruleDefinition

JPMorganPaymentCallbackEvent dataclass

JPMorganPaymentCallbackEvent(
    end_to_end_id,
    firm_root_id,
    payment_status,
    create_date_time,
    exceptions,
    raw=None,
)

Doc -> https://developer.payments.jpmorgan.com/docs/treasury/global-payments/capabilities/global-payments/callbacks ⧉ Example payload: { "callbacks": [ { "endToEndId": "MY20230727_0219", "createDateTime": "2023-07-27T07:41:54.773Z", "paymentStatus": "REJECTED", "firmRootId": "53038faa-43f6-42fc-9545-648fd2dab411", "exceptions": [ { "errorCode": "10004", "errorDescription": "Error occurred on /requestedExecutionDate", "ruleDefinition": "Date validation failure" } ] } ] }

create_date_time instance-attribute

create_date_time

end_to_end_id instance-attribute

end_to_end_id

exceptions instance-attribute

exceptions

firm_root_id instance-attribute

firm_root_id

from_dict staticmethod

from_dict(data)

Source code in shared/services/payment_providers/jpmorgan/entities.py

@staticmethod
def from_dict(data: dict) -> "JPMorganPaymentCallbackEvent":  # type: ignore[type-arg]
    exceptions_data = data.get("exceptions", [])
    exceptions = [
        JPMorganCallbackException(
            errorCode=ex["errorCode"],
            errorDescription=ex["errorDescription"],
            ruleDefinition=ex["ruleDefinition"],
        )
        for ex in exceptions_data
    ]
    return JPMorganPaymentCallbackEvent(
        end_to_end_id=data["endToEndId"],
        firm_root_id=data["firmRootId"],
        payment_status=JPMorganTransactionState(data["paymentStatus"]),
        create_date_time=data["createDateTime"],
        exceptions=exceptions,
        raw=data,
    )

payment_status instance-attribute

payment_status

raw class-attribute instance-attribute

raw = None

JPMorganTransactionState

Bases: AlanBaseEnum

blocked class-attribute instance-attribute

blocked = 'BLOCKED'

Payment blocked due to sanctions issue

completed class-attribute instance-attribute

completed = 'COMPLETED'

Payment has successfully completed

completed_credited class-attribute instance-attribute

completed_credited = 'COMPLETED_CREDITED'

Status indicating the beneficiary's account has been credited

pending class-attribute instance-attribute

pending = 'PENDING'

Payment is pending processing

pending_posting class-attribute instance-attribute

pending_posting = 'PENDING_POSTING'

Payment is yet to be posted in the beneficiary account

rejected class-attribute instance-attribute

rejected = 'REJECTED'

Payment has been rejected. Please refer to the exception object for error details

returned class-attribute instance-attribute

returned = 'RETURNED'

Payment has been retured to the debtor party

warehoused class-attribute instance-attribute

warehoused = 'WAREHOUSED'

Payment request was successfully received. The request will be processed in the next available window, typically the next calendar day

shared.services.payment_providers.jpmorgan.errors

JPMorganBadRequestError

JPMorganBadRequestError(code=None, message=None)

Bases: JPMorganError

Our request was not formatted correctly, maybe their API has evolved or we have missed something

List of Error codes and Rule definitions.

Error Code	Rule Definition
10001	Mandatory field is missing or invalid
10002	Minimum length validation failure
10003	Maximum length validation failure
10004	Date validation failure
10005	Amount validation failure ~ value more than maximum
10006	Amount validation failure ~ value less than minimum
10007	Amount validation failure ~ value is not a number
10008	Validation failure ~ unexpected value provided
10009	Invalid Id provided
10010	Personal information validation failure
12000	System error
13000	Uncategorized error

errorDescription is dynamically generated, hence not shown here.

Standard API Gateway Error codes and descriptions

Error Code	Description
GCA-023	Please re-send request in valid format
GCA-030	API Processing Error
GCA-148	debtor Account id must be provided
GCA-149	debtorAgent bic or clearingSystemId must be provided
GCA-150	debtor account id/bic was not found
GCA-154	Mandatory field paymentType is invalid or missing

Source code in shared/services/payment_providers/jpmorgan/errors.py

def __init__(self, code: str | None = None, message: str | None = None) -> None:
    self.code = code
    self.message = message

JPMorganError

JPMorganError(code=None, message=None)

Bases: Exception

A generic JPMorgan exception we should handle

Source code in shared/services/payment_providers/jpmorgan/errors.py

def __init__(self, code: str | None = None, message: str | None = None) -> None:
    self.code = code
    self.message = message

code instance-attribute

code = code

message instance-attribute

message = message

JPMorganForbiddenError

JPMorganForbiddenError(code=None, message=None)

Bases: JPMorganError

We get a 403, for one of the below reasons | Error Code | Description | |----------- |-------------------------------------------| | GCA-001 |Client is not eligible for the API Service | | GCA-003 |Client is not eligible for the API Service | | GCA-145 |incorrect originator account id provided | | GCA-150 |debtor account id/bic was not found |

Source code in shared/services/payment_providers/jpmorgan/errors.py

def __init__(self, code: str | None = None, message: str | None = None) -> None:
    self.code = code
    self.message = message

JPMorganInternalError

JPMorganInternalError(code=None, message=None)

Bases: JPMorganError

JPMorgan issue, we can't act on it

Source code in shared/services/payment_providers/jpmorgan/errors.py

def __init__(self, code: str | None = None, message: str | None = None) -> None:
    self.code = code
    self.message = message

JPMorganNotFoundError

JPMorganNotFoundError(code=None, message=None)

Bases: JPMorganError

Resource does not exist on JPMorgan, it is up to us to decide whether we take this error into account or not

Source code in shared/services/payment_providers/jpmorgan/errors.py

def __init__(self, code: str | None = None, message: str | None = None) -> None:
    self.code = code
    self.message = message

shared.services.payment_providers.jpmorgan.helpers

get_jpmorgan_config_from_business_account_name

get_jpmorgan_config_from_business_account_name(
    account_name,
)

Source code in shared/services/payment_providers/jpmorgan/helpers.py

def get_jpmorgan_config_from_business_account_name(
    account_name: JPMorganBusinessAccountName,
) -> JPMorganBusinessAccountConfig:
    if JPMorganBusinessAccountName[account_name] not in JPMorganBusinessAccountName:
        raise ValueError(
            f"Invalid JPMorgan business account name {account_name}. "
            f"Valid values are: {', '.join(JPMorganBusinessAccountName.get_values())}"
        )

    jpmorgan_config = _business_account_name_mapping[account_name]

    if not jpmorgan_config:
        raise ValueError(
            f"Could not find JPMorgan config for business account name {account_name}"
        )

    return jpmorgan_config

shared.services.payment_providers.jpmorgan.jpmorgan_client

JPMorganClient

JPMorganClient(business_account_name)

See the authentication documentation here: https://developer.jpmorgan.com/products/tsapi-onboarding-guides/guides/digitally-sign-payment-requests-(required ⧉)

And the payment endpoint here: https://developer.jpmorgan.com/products/global-rtp/specification ⧉

Source code in shared/services/payment_providers/jpmorgan/jpmorgan_client.py

def __init__(
    self,
    business_account_name: JPMorganBusinessAccountName,
) -> None:
    jpmorgan_config = get_jpmorgan_config_from_business_account_name(
        business_account_name
    )

    signature_private_key = raw_secret_from_config(
        config_key=jpmorgan_config.signature_private_key_secret_name,
        default_secret_value=current_config.get(
            jpmorgan_config.signature_private_key
        ),
    )
    if signature_private_key is None:
        raise ValueError("JPMorgan signature private key is not configured")

    transport_private_key = raw_secret_from_config(
        config_key=jpmorgan_config.transport_private_key_secret_name,
        default_secret_value=current_config.get(
            jpmorgan_config.transport_private_key
        ),
    )
    if transport_private_key is None:
        raise ValueError("JPMorgan transport private key is not configured")

    transport_certificate = raw_secret_from_config(
        config_key=jpmorgan_config.transport_public_certificate_secret_name,
        default_secret_value=current_config.get(
            jpmorgan_config.transport_public_certificate
        ),
    )
    if transport_certificate is None:
        raise ValueError("JPMorgan transport public certificate is not configured")

    self._signature_private_key = raw_secret_from_config(
        config_key=jpmorgan_config.signature_private_key_secret_name,
        default_secret_value=current_config.get(
            jpmorgan_config.signature_private_key
        ),
    )
    if self._signature_private_key is None:
        raise ValueError("JPMorgan signature private key is not configured")

    self._base_url = get_config_or_fail(jpmorgan_config.base_url)
    self._account_id = get_config_or_fail(jpmorgan_config.account_id)
    self._debtor_agent_bic = get_config_or_fail(jpmorgan_config.debtor_agent_bic)

    # To send client certificates with our request, we need to pass the requests lib (in fact, the
    # underlying OpenSSL implementation) paths to the files, not the content itself. That's why we
    # need to create named temporary files here, so we can reference them later when sending the
    # Payment.
    # In Python 3.12, we should use delete_on_close=False instead
    self._transport_private_key_temp_file = tempfile.NamedTemporaryFile(
        delete=False, suffix=".key"
    )
    self._transport_private_key_temp_file.write(str.encode(transport_private_key))
    self._transport_private_key_temp_file.close()

    # In Python 3.12, we should use delete_on_close=False instead
    self._transport_cert_temp_file = tempfile.NamedTemporaryFile(
        delete=False, suffix=".cert"
    )
    self._transport_cert_temp_file.write(str.encode(transport_certificate))
    self._transport_cert_temp_file.close()

PAYMENTS_ENDPOINT class-attribute instance-attribute

PAYMENTS_ENDPOINT = 'payments'

get_supported_currency

get_supported_currency()

Source code in shared/services/payment_providers/jpmorgan/jpmorgan_client.py

def get_supported_currency(self) -> str:
    return current_config.get("JPMORGAN_CURRENCY", "EUR")  # type: ignore[no-any-return]

get_transaction_info

get_transaction_info(
    *,
    transaction_id=None,
    request_id=None,
    ignore_not_found_error=False
)

https://developer.payments.jpmorgan.com/api/treasury/global-payments/global-payments/global-payments#/operations/getPaymentDetails ⧉

Source code in shared/services/payment_providers/jpmorgan/jpmorgan_client.py

def get_transaction_info(  # type: ignore[no-untyped-def]
    self,
    *,
    transaction_id: str | None = None,
    request_id: str | None = None,
    ignore_not_found_error: bool = False,
):
    """
    https://developer.payments.jpmorgan.com/api/treasury/global-payments/global-payments/global-payments#/operations/getPaymentDetails
    """

    if transaction_id is None and request_id is None:
        raise ValueError("Either transaction_id or request_id must be provided")
    if transaction_id and request_id:
        raise ValueError(
            "Only one of transaction_id or request_id must be provided"
        )

    request = requests.Request(
        method="GET",
        url=self._base_url + self.PAYMENTS_ENDPOINT,
        params=(
            {"firmRootId": transaction_id}
            if transaction_id
            else {"endToEndId": request_id}
        ),
        headers=self._get_header(),
    ).prepare()

    current_logger.debug(
        "Calling JPMorgan API: GET %s",
        request.url,
        jpmorgan={
            "transaction_id": transaction_id,
            "request_id": request_id,
        },
    )

    response = requests.Session().send(
        request,
        cert=self._get_transport_certificate(),
    )

    try:
        self._raise_for_status(response)
    except JPMorganError as e:
        if ignore_not_found_error and isinstance(e, JPMorganNotFoundError):
            return None
        raise e

    return response.json()

pay

pay(
    request_id,
    amount_in_cents,
    ibancode,
    swift=None,
    account_type=JPMorganAccountType.IBAN,
    payment_description=None,
    reimbursement_payment_id=None,
    recipient_first_name=None,
    recipient_last_name=None,
    recipient_company_name=None,
    address=None,
)

https://developer.payments.jpmorgan.com/api/treasury/global-payments/global-payments/global-payments#/operations/initiatePayments ⧉

Trigger a payment using JPMorgan Treasury API.

The payment will be sent to the recipient described by the recipient_* fields.

Source code in shared/services/payment_providers/jpmorgan/jpmorgan_client.py

def pay(
    self,
    # The request_id is controlled by us, and is a string of 35 chars max, letters and numbers only
    request_id: str,
    amount_in_cents: int,
    ibancode: str,
    swift: str | None = None,  # SWIFT / BIC is not mandatory
    account_type: JPMorganAccountType = JPMorganAccountType.IBAN,
    payment_description: str | None = None,
    reimbursement_payment_id: str | None = None,  # noqa: ARG002
    # These fields must be specified if the recipient is an individual
    recipient_first_name: str | None = None,
    recipient_last_name: str | None = None,
    # This field must be specified if the recipient is a company or a hospital
    recipient_company_name: str | None = None,
    address: BillingAddress | None = None,
) -> PaymentInfo:
    """
    https://developer.payments.jpmorgan.com/api/treasury/global-payments/global-payments/global-payments#/operations/initiatePayments

    Trigger a payment using JPMorgan Treasury API.

    The payment will be sent to the recipient described by the recipient_* fields.
    """
    if not isinstance(amount_in_cents, int):
        raise ValueError("amount is not int")

    if not request_id:
        raise ValueError("request_id is not set")

    if recipient_company_name:
        creditor_name = f"{recipient_company_name}"
    else:
        creditor_name = f"{recipient_first_name} {recipient_last_name}"

    payment_currency = self.get_supported_currency()

    # SEPA Instant is supported on - Frankfurt (CHASDEFX); Dublin (CHASIE4L); Luxembourg (CHASLULX)
    is_sepa_instant = self._debtor_agent_bic in ["CHASDEFX", "CHASIE4L", "CHASLULX"]

    self._check_account_type_requirements(
        account_type=account_type,
        is_sepa_instant=is_sepa_instant,
        debtor_agent_bic=self._debtor_agent_bic,
        address=address,
        payment_currency=payment_currency,
    )

    payload = {
        "payments": {
            "requestedExecutionDate": date.today().strftime("%Y-%m-%d"),
            "paymentIdentifiers": {"endToEndId": request_id},
            "paymentCurrency": payment_currency,
            "paymentAmount": round(
                amount_in_cents / 100,
                2,  # Only 2 decimal is allowed, safety measure
            ),
            "debtor": {
                "debtorName": "Alan Insurance",
                "debtorAccount": {
                    "accountId": self._account_id,
                },
            },
            "debtorAgent": {
                "financialInstitutionId": {"bic": self._debtor_agent_bic}
            },
            "creditor": {
                "creditorAccount": {
                    "accountId": ibancode,
                    "accountType": str(account_type),
                },
                "creditorName": creditor_name[:140],  # 140 chars max
            },
            "transferType": "CREDIT",
        }
    }

    if is_sepa_instant:
        # Mandatory values only for SEPA INSTANT
        payload["payments"]["paymentType"] = "RTP"
        payload["payments"]["debtor"]["debtorAccount"]["accountCurrency"] = "EUR"  # type: ignore[index]
        payload["payments"]["debtor"]["debtorAccount"]["accountType"] = "IBAN"  # type: ignore[index]

    if swift:
        payload["payments"]["creditorAgent"] = {
            "financialInstitutionId": {"bic": swift}
        }

    if address:
        payload["payments"]["creditor"]["postalAddress"] = {  # type: ignore[index]
            "addressType": "ADDR",  # "Postal Address is the complete postal address."
            # ruff: noqa: ERA001
            # If client elects to use the structured address fields populate Address Line + Country
            # 2 lines, 70 characters each including spaces
            # Note: structured address fields ((Postal Address fields except Address Line) + Town Name + Country)
            # will be mandatory from November 2026, this will mean unstructured address fields will no longer be
            # supported (Address Line) past this date
            # -> https://linear.app/alan-eu/issue/FREF-390/jpmorgan-api-address-requirements-before-112026
            # -- Structured fields:
            # "streetName: f"{address.street}"[:140],
            # "buildingNumber": f"{address.street}"[:140], # We should extract the number from here
            # "postalCode": f"{address.postal_code}"[:140],
            # "townName": f"{address.city}"[:140],
            # -- Unstructured fields:
            "addressLine": [
                f"{address.street}"[:70],
                f"{address.postal_code} {address.city}"[:70],
            ],
            "country": address.country,  # ISO Code needed here
        }
        payload["payments"]["creditor"][  # type: ignore[index]
            "countryOfResidence"
        ] = address.country  # ISO Code needed here

    if payment_description:
        payload["payments"]["remittanceInformation"] = {
            "unstructuredInformation": [payment_description[:140]]
        }

    signed_payload = self._build_jwt(payload)

    request = requests.Request(
        method="POST",
        url=self._base_url + self.PAYMENTS_ENDPOINT,
        headers=self._get_header(),
        data=signed_payload,
    ).prepare()

    current_logger.debug(
        "Calling JPMorgan API: POST %s",
        request.url,
        jpmorgan={
            "request_id": request_id,
        },
    )

    # No advisory lock here, we assume we can create simultaneous payments on the JPMorgan API
    response = requests.Session().send(
        request,
        cert=self._get_transport_certificate(),
    )

    self._raise_for_status(response)

    response_json = response.json()

    firm_root_id = response_json["paymentInitiationResponse"]["firmRootId"]
    created_at = datetime.datetime.utcnow()

    return PaymentInfo(
        transaction_id=firm_root_id,
        request_id=request_id,
        created_at=created_at,
        currency=payment_currency,
        payment_payload=payload,
    )

shared.services.payment_providers.jpmorgan.noop_jpmorgan_client

NoopJPMorganClient

JPMorgan client that does nothing

get_supported_currency

get_supported_currency()

Source code in shared/services/payment_providers/jpmorgan/noop_jpmorgan_client.py

def get_supported_currency(self) -> str:
    from shared.helpers.app_name import AppName, get_current_app_name

    current_app = get_current_app_name()
    if current_app == AppName.ALAN_CA:
        return "CAD"
    return "EUR"

pay

pay(
    request_id,
    amount_in_cents,
    ibancode,
    swift=None,
    account_type=JPMorganAccountType.IBAN,
    payment_description=None,
    reimbursement_payment_id=None,
    recipient_first_name=None,
    recipient_last_name=None,
    recipient_company_name=None,
    address=None,
)

Source code in shared/services/payment_providers/jpmorgan/noop_jpmorgan_client.py

def pay(
    self,
    # The request_id is controlled by us, and is a string of 35 chars max, letters and numbers only
    request_id: str,
    amount_in_cents: int,
    ibancode: str,  # noqa: ARG002
    swift: str | None = None,  # noqa: ARG002
    account_type: JPMorganAccountType = JPMorganAccountType.IBAN,
    payment_description: str | None = None,  # noqa: ARG002
    reimbursement_payment_id: str | None = None,  # noqa: ARG002
    # These fields must be specified if the recipient is an individual
    recipient_first_name: str | None = None,  # noqa: ARG002
    recipient_last_name: str | None = None,  # noqa: ARG002
    # This field must be specified if the recipient is a company or a hospital
    recipient_company_name: str | None = None,  # noqa: ARG002
    address: BillingAddress | None = None,
) -> PaymentInfo:
    if not isinstance(amount_in_cents, int):
        raise ValueError("amount is not int")

    if not request_id:
        raise ValueError("request_id is not set")

    currency = self.get_supported_currency()

    if currency == "CAD":
        JPMorganClient._check_account_type_requirements(  # noqa: ALN027, for testing purposes
            account_type=account_type,
            is_sepa_instant=False,
            debtor_agent_bic="CHASCATT",
            address=address,
            payment_currency=currency,
        )

    # We generate a fake `payment_id`
    return PaymentInfo(
        transaction_id=str(uuid.uuid4()),
        request_id=request_id,
        created_at=datetime.now(),
        currency=currency,
    )