Skip to content

Adapters

components.payment_gateway.subcomponents.accounts.adapters.account_balance_adapter

Adapter contract for fetching account balances from PSPs.

Each PSP implementation is responsible for fan-out: JPM returns all entitled accounts in a single call, Revolut needs one call per currency, etc. The caller is handed back a flat list of AccountBalance and decides what to do with it.

AccountBalance dataclass 

AccountBalance(id, name, available_amount, currency, as_of)

Bases: DataClassJsonMixin

A balance snapshot for a single PSP account.

as_of instance-attribute 

as_of

Timestamp at which the PSP captured this snapshot.

available_amount instance-attribute 

available_amount

Amount in minor currency units (e.g. cents for CAD).

currency instance-attribute 

currency

ISO 4217 currency code (e.g. "CAD").

id instance-attribute 

id

ID of the account

name instance-attribute 

name

Name of the account at PSP level.

AccountBalanceAdapter

Bases: ABC

Adapter for PSPs that expose account balances.

list_account_balances abstractmethod 

list_account_balances(*, workspace_key)
Source code in components/payment_gateway/subcomponents/accounts/adapters/account_balance_adapter.py 
40
41
@abstractmethod
def list_account_balances(self, *, workspace_key: str) -> list[AccountBalance]: ...

components.payment_gateway.subcomponents.accounts.adapters.account_balance_adapter_registry

AccountBalanceAdapterRegistry 

AccountBalanceAdapterRegistry(adapters)

Registry of adapters that expose account balances per PSP.

Source code in components/payment_gateway/subcomponents/accounts/adapters/account_balance_adapter_registry.py 
70
71
72
73
def __init__(
    self, adapters: dict[PaymentServiceProvider, AccountBalanceAdapter]
) -> None:
    self.adapters = adapters

adapters instance-attribute 

adapters = adapters

create classmethod 

create()

Build adapters for every PSP available in the current environment.

Source code in components/payment_gateway/subcomponents/accounts/adapters/account_balance_adapter_registry.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
@classmethod
def create(cls) -> "AccountBalanceAdapterRegistry":
    """Build adapters for every PSP available in the current environment."""
    adapters: dict[PaymentServiceProvider, AccountBalanceAdapter] = {}
    try:
        adapters[PaymentServiceProvider.jpmorgan] = (
            JPMorganAccountBalanceAdapter.create()
        )
        adapters[PaymentServiceProvider.revolut] = (
            RevolutAccountBalanceAdapter.create()
        )
    except Exception:  # noqa: S110
        # JPMorgan not available in this environment
        pass

    return AccountBalanceAdapterRegistry(adapters=adapters)

create_null classmethod 

create_null(
    *,
    track_jpmorgan_requests=None,
    jpmorgan_responses=None,
    track_revolut_requests=None,
    revolut_responses=None,
    revolut_simulator=None
)

Null registry for tests.

Source code in components/payment_gateway/subcomponents/accounts/adapters/account_balance_adapter_registry.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
@classmethod
def create_null(
    cls,
    *,
    track_jpmorgan_requests: list[JPMorganTrackedRequest] | None = None,
    jpmorgan_responses: list[JPMorganConfiguredResponse] | None = None,
    track_revolut_requests: list[RevolutTrackedRequest] | None = None,
    revolut_responses: list[RevolutConfiguredResponse] | None = None,
    revolut_simulator: RevolutBusinessSimulator | None = None,
) -> "AccountBalanceAdapterRegistry":
    """Null registry for tests."""
    return AccountBalanceAdapterRegistry(
        adapters={
            PaymentServiceProvider.jpmorgan: JPMorganAccountBalanceAdapter.create_null(
                track_requests=track_jpmorgan_requests,
                responses=jpmorgan_responses,
            ),
            PaymentServiceProvider.revolut: RevolutAccountBalanceAdapter.create_null(
                track_requests=track_revolut_requests,
                responses=revolut_responses,
                simulator=revolut_simulator,
            ),
        }
    )

get_account_balance_adapter 

get_account_balance_adapter(provider)
Source code in components/payment_gateway/subcomponents/accounts/adapters/account_balance_adapter_registry.py 
75
76
77
78
79
80
81
82
83
84
def get_account_balance_adapter(
    self, provider: PaymentServiceProvider
) -> AccountBalanceAdapter:
    adapter = self.adapters.get(provider)
    if adapter is None:
        raise ProviderNotSupportedException(
            provider,
            f"Account balance lookup is not supported by provider {provider}",
        )
    return adapter

components.payment_gateway.subcomponents.accounts.adapters.adyen

adyen_provider_account_adapter

AdyenProviderAccountAdapter 

AdyenProviderAccountAdapter(balance_accounts_client)

Bases: ProviderAccountAdapter

Adapter for fetching a single Adyen balance account.

Adyen exposes a balance account directly by its id (BA…) via the Balance Platform Configuration API. Amounts are already in minor units, so they are passed through unchanged.

Tags
Source code in components/payment_gateway/subcomponents/accounts/adapters/adyen/adyen_provider_account_adapter.py 
62
63
64
65
66
def __init__(
    self,
    balance_accounts_client: AdyenBalanceAccountsApiClient,
) -> None:
    self._balance_accounts_client = balance_accounts_client
create classmethod 
create()

Normal factory.

Source code in components/payment_gateway/subcomponents/accounts/adapters/adyen/adyen_provider_account_adapter.py 
38
39
40
41
42
43
44
45
@classmethod
def create(cls) -> "AdyenProviderAccountAdapter":
    """Normal factory."""
    return cls(
        balance_accounts_client=AdyenBalanceAccountsApiClient.create(
            credentials=get_adyen_balance_accounts_api_client_credentials()
        )
    )
create_null classmethod 
create_null(*, track_requests=None, responses=None)

Null factory.

Source code in components/payment_gateway/subcomponents/accounts/adapters/adyen/adyen_provider_account_adapter.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
@classmethod
def create_null(
    cls,
    *,
    track_requests: list[tuple[str, dict, dict]] | None = None,  # type: ignore[type-arg]
    responses: list[tuple[int, dict]] | None = None,  # type: ignore[type-arg]
) -> "AdyenProviderAccountAdapter":
    """Null factory."""
    return cls(
        balance_accounts_client=AdyenBalanceAccountsApiClient.create_null(
            track_requests=track_requests,
            responses=responses,
        )
    )
get_account 
get_account(*, external_id)
Source code in components/payment_gateway/subcomponents/accounts/adapters/adyen/adyen_provider_account_adapter.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
@override
def get_account(self, *, external_id: str) -> ProviderAccount:
    from Adyen.exceptions import AdyenError

    try:
        balance_account = self._balance_accounts_client.get_balance_account(
            id=external_id
        )
    except AdyenError as error:
        if str(error.error_code) == _BALANCE_ACCOUNT_NOT_FOUND_ERROR_CODE:
            raise AccountNotFoundOnProviderException(external_id) from error
        raise

    as_of = datetime.now(UTC)
    return ProviderAccount(
        id=balance_account.id,
        name=balance_account.description,
        status=balance_account.status,
        balances=[
            AccountBalance(
                id=balance_account.id,
                name=balance_account.description,
                available_amount=balance.available,
                currency=balance.currency,
                as_of=as_of,
            )
            for balance in balance_account.balances or []
        ],
    )

helpers

to_balance_account_info 

to_balance_account_info(
    external_account_holder_id, description, reference=None
)

Convert our account model into an Adyen BalanceAccountInfo for API creation calls.

Source code in components/payment_gateway/subcomponents/accounts/adapters/adyen/helpers.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
def to_balance_account_info(
    external_account_holder_id: str,
    description: str,
    reference: str | None = None,
) -> AdyenBalanceAccountInfo:
    """
    Convert our account model into an Adyen BalanceAccountInfo for API creation calls.
    """

    return AdyenBalanceAccountInfo(
        accountHolderId=external_account_holder_id,
        description=description[0:300],
        reference=reference[0:150] if reference else None,
    )

components.payment_gateway.subcomponents.accounts.adapters.jpmorgan

jpmorgan_account_balance_adapter

JPMorganAccountBalanceAdapter 

JPMorganAccountBalanceAdapter(jpmorgan_client_factory)

Bases: AccountBalanceAdapter

Source code in components/payment_gateway/subcomponents/accounts/adapters/jpmorgan/jpmorgan_account_balance_adapter.py 
53
54
55
56
57
58
59
60
61
62
def __init__(
    self,
    jpmorgan_client_factory: Callable[
        [JPMorganBusinessAccountName], JPMorganAccountBalancesApiClient
    ],
) -> None:
    self._jpmorgan_client_factory = jpmorgan_client_factory
    self._client_cache: dict[
        JPMorganBusinessAccountName, JPMorganAccountBalancesApiClient
    ] = {}
create classmethod 
create()
Source code in components/payment_gateway/subcomponents/accounts/adapters/jpmorgan/jpmorgan_account_balance_adapter.py 
28
29
30
31
32
33
34
35
36
37
38
@classmethod
def create(cls) -> "JPMorganAccountBalanceAdapter":
    return cls(
        jpmorgan_client_factory=lambda business_account_name: (
            JPMorganAccountBalancesApiClient.create(
                credentials=get_jpmorgan_account_balances_api_client_credentials(
                    business_account_name=business_account_name
                ),
            )
        ),
    )
create_null classmethod 
create_null(*, track_requests=None, responses=None)
Source code in components/payment_gateway/subcomponents/accounts/adapters/jpmorgan/jpmorgan_account_balance_adapter.py 
40
41
42
43
44
45
46
47
48
49
50
51
@classmethod
def create_null(
    cls,
    *,
    track_requests: list[TrackedRequest] | None = None,
    responses: list[ConfiguredResponse] | None = None,
) -> "JPMorganAccountBalanceAdapter":
    jpmorgan_client = JPMorganAccountBalancesApiClient.create_null(
        track_requests=track_requests,
        responses=responses,
    )
    return cls(jpmorgan_client_factory=lambda _: jpmorgan_client)
list_account_balances 
list_account_balances(*, workspace_key)
Source code in components/payment_gateway/subcomponents/accounts/adapters/jpmorgan/jpmorgan_account_balance_adapter.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
@override
def list_account_balances(self, *, workspace_key: str) -> list[AccountBalance]:
    jpmorgan_client = self._get_jpmorgan_client(workspace_key)
    response = jpmorgan_client.get_account_balances()
    balances: list[AccountBalance] = []
    for account in response.accountList or []:
        currency = account.currency
        if currency is None or currency.code is None:
            continue
        currency_code = currency.code
        decimal_location = (
            int(currency.decimalLocation)
            if currency.decimalLocation is not None
            else 2
        )
        multiplier = 10**decimal_location
        for balance in account.balanceList or []:
            if balance.openingAvailableAmount is None:
                continue
            balances.append(
                AccountBalance(
                    id=account.accountId,
                    name=account.accountName,
                    available_amount=float_amount_to_minor_units(
                        balance.openingAvailableAmount, multiplier=multiplier
                    ),
                    currency=currency_code,
                    as_of=_parse_record_timestamp(balance.recordTimestamp),
                )
            )
    return balances

components.payment_gateway.subcomponents.accounts.adapters.provider_account_adapter

Adapter contract for fetching a single Provider Account, with its balances.

A Provider Account (AKA PSP Account) is a virtual account (not a real Bank Account with an IBAN and KYC and stuff) that can hold a balance. Transfers can typically occur between accounts on the same PSP without fees or delays. Usually not accessible from outside the PSP, although some can function as Bank Accounts if properly KYC'd. Example: an Adyen balance account (BA...), mapped to an Account via its external_id.

Business context — Alan Flex (🇪🇸): Flex is Alan's flexible-benefits product. Each client company gets its own Provider Account (today an Adyen balance account) that the company pre-funds; that balance backs its employees' Flex card spending.

ProviderAccount dataclass 

ProviderAccount(id, name, status, balances)

Bases: DataClassJsonMixin

A single provider account, with the balances it currently holds.

balances instance-attribute 

balances

Balances held by the account, one entry per currency.

id instance-attribute 

id

ID of the account at PSP level.

name instance-attribute 

name

Name of the account at PSP level.

status instance-attribute 

status

Status of the account at PSP level (e.g. "active").

ProviderAccountAdapter

Bases: ABC

Adapter for PSPs that expose a single account by its PSP id.

get_account abstractmethod 

get_account(*, external_id)
Source code in components/payment_gateway/subcomponents/accounts/adapters/provider_account_adapter.py 
44
45
@abstractmethod
def get_account(self, *, external_id: str) -> ProviderAccount: ...

components.payment_gateway.subcomponents.accounts.adapters.provider_account_adapter_registry

ProviderAccountAdapterRegistry 

ProviderAccountAdapterRegistry(adapters)

Registry of adapters that fetch a single account by its PSP id.

Source code in components/payment_gateway/subcomponents/accounts/adapters/provider_account_adapter_registry.py 
45
46
47
48
def __init__(
    self, adapters: dict[PaymentServiceProvider, ProviderAccountAdapter]
) -> None:
    self.adapters = adapters

adapters instance-attribute 

adapters = adapters

create classmethod 

create()

Build adapters for every PSP available in the current environment.

Source code in components/payment_gateway/subcomponents/accounts/adapters/provider_account_adapter_registry.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
@classmethod
def create(cls) -> "ProviderAccountAdapterRegistry":
    """Build adapters for every PSP available in the current environment."""
    adapters: dict[PaymentServiceProvider, ProviderAccountAdapter] = {}
    try:
        adapters[PaymentServiceProvider.adyen] = (
            AdyenProviderAccountAdapter.create()
        )
    except Exception:  # noqa: S110
        # Adyen not available in this environment
        pass

    return ProviderAccountAdapterRegistry(adapters=adapters)

create_null classmethod 

create_null(*, track_requests=None, responses=None)

Null registry for tests.

Source code in components/payment_gateway/subcomponents/accounts/adapters/provider_account_adapter_registry.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
@classmethod
def create_null(
    cls,
    *,
    track_requests: list[tuple[str, dict, dict]] | None = None,  # type: ignore[type-arg]
    responses: list[tuple[int, dict]] | None = None,  # type: ignore[type-arg]
) -> "ProviderAccountAdapterRegistry":
    """Null registry for tests."""
    return ProviderAccountAdapterRegistry(
        adapters={
            PaymentServiceProvider.adyen: AdyenProviderAccountAdapter.create_null(
                track_requests=track_requests,
                responses=responses,
            ),
        }
    )

get_account_adapter 

get_account_adapter(provider)
Source code in components/payment_gateway/subcomponents/accounts/adapters/provider_account_adapter_registry.py 
50
51
52
53
54
55
56
57
58
59
def get_account_adapter(
    self, provider: PaymentServiceProvider
) -> ProviderAccountAdapter:
    adapter = self.adapters.get(provider)
    if adapter is None:
        raise ProviderNotSupportedException(
            provider,
            f"Single-account lookup is not supported by provider {provider}",
        )
    return adapter

components.payment_gateway.subcomponents.accounts.adapters.revolut

revolut_account_balance_adapter

RevolutAccountBalanceAdapter 

RevolutAccountBalanceAdapter(revolut_client_factory)

Bases: AccountBalanceAdapter

Fetches account balances from Revolut.

Revolut exposes one account per currency for a given business account, so a single get_accounts call returns the balances for all of them. A dedicated RevolutBusinessApiClient is resolved (and cached) per workspace, since each workspace maps to its own Revolut business account.

Source code in components/payment_gateway/subcomponents/accounts/adapters/revolut/revolut_account_balance_adapter.py 
63
64
65
66
67
68
69
70
71
72
def __init__(
    self,
    revolut_client_factory: Callable[
        [RevolutBusinessAccountName], RevolutBusinessApiClient
    ],
) -> None:
    self._revolut_client_factory = revolut_client_factory
    self._client_cache: dict[
        RevolutBusinessAccountName, RevolutBusinessApiClient
    ] = {}
create classmethod 
create()
Source code in components/payment_gateway/subcomponents/accounts/adapters/revolut/revolut_account_balance_adapter.py 
38
39
40
41
42
43
44
45
46
47
48
@classmethod
def create(cls) -> "RevolutAccountBalanceAdapter":
    return cls(
        revolut_client_factory=lambda business_account_name: (
            RevolutBusinessApiClient.create(
                credentials=get_revolut_business_api_client_credentials(
                    business_account_name=business_account_name
                ),
            )
        )
    )
create_null classmethod 
create_null(
    *, track_requests=None, responses=None, simulator=None
)
Source code in components/payment_gateway/subcomponents/accounts/adapters/revolut/revolut_account_balance_adapter.py 
50
51
52
53
54
55
56
57
58
59
60
61
@classmethod
def create_null(
    cls,
    *,
    track_requests: list[TrackedRequest] | None = None,
    responses: list[ConfiguredResponse] | None = None,
    simulator: RevolutBusinessSimulator | None = None,
) -> "RevolutAccountBalanceAdapter":
    revolut_client = RevolutBusinessApiClient.create_null(
        track_requests=track_requests, responses=responses, simulator=simulator
    )
    return cls(revolut_client_factory=lambda _: revolut_client)
list_account_balances 
list_account_balances(*, workspace_key)
Source code in components/payment_gateway/subcomponents/accounts/adapters/revolut/revolut_account_balance_adapter.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
@override
def list_account_balances(self, *, workspace_key: str) -> list[AccountBalance]:
    revolut_client = self._get_revolut_client(workspace_key)
    balances = []

    for account in revolut_client.get_accounts():
        balances.append(
            AccountBalance(
                id=str(account.id),
                name=account.name,
                available_amount=float_amount_to_minor_units(
                    account.balance, multiplier=100
                ),
                currency=account.currency.root,
                as_of=account.updated_at,
            )
        )

    return balances