Skip to content

Queries

components.payment_gateway.subcomponents.accounts.protected.business_logic.queries.account_balance_queries

AccountBalanceQueries 

AccountBalanceQueries(adapter_registry)

Public queries for fetching account balances across PSPs.

Source code in components/payment_gateway/subcomponents/accounts/protected/business_logic/queries/account_balance_queries.py 
54
55
def __init__(self, adapter_registry: AccountBalanceAdapterRegistry) -> None:
    self.adapter_registry = adapter_registry

adapter_registry instance-attribute 

adapter_registry = adapter_registry

create classmethod 

create()

Normal factory.

Source code in components/payment_gateway/subcomponents/accounts/protected/business_logic/queries/account_balance_queries.py 
28
29
30
31
@classmethod
def create(cls) -> "AccountBalanceQueries":
    """Normal factory."""
    return cls(adapter_registry=AccountBalanceAdapterRegistry.create())

create_null classmethod 

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

Null factory for tests.

Source code in components/payment_gateway/subcomponents/accounts/protected/business_logic/queries/account_balance_queries.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@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,
) -> "AccountBalanceQueries":
    """Null factory for tests."""
    return cls(
        adapter_registry=AccountBalanceAdapterRegistry.create_null(
            track_jpmorgan_requests=track_jpmorgan_requests,
            jpmorgan_responses=jpmorgan_responses,
            track_revolut_requests=track_revolut_requests,
            revolut_responses=revolut_responses,
            revolut_simulator=revolut_simulator,
        )
    )

list_account_balances 

list_account_balances(*, workspace_key)

Return all balances exposed by the PSP backing workspace_key.

Source code in components/payment_gateway/subcomponents/accounts/protected/business_logic/queries/account_balance_queries.py 
57
58
59
60
61
62
63
64
@obs.api_call()
def list_account_balances(self, *, workspace_key: str) -> list[AccountBalance]:
    """Return all balances exposed by the PSP backing ``workspace_key``."""
    provider = get_provider_for_workspace(workspace_key)
    if provider is None:
        raise WorkspaceNotRegisteredException(workspace_key)
    adapter = self.adapter_registry.get_account_balance_adapter(provider)
    return adapter.list_account_balances(workspace_key=workspace_key)

components.payment_gateway.subcomponents.accounts.protected.business_logic.queries.account_holder_queries

AccountHolderQueries

This class contains all the queries related to account holders.

get_account_holder 

get_account_holder(session, /, id)

Get an account holder entity from its ID.

Source code in components/payment_gateway/subcomponents/accounts/protected/business_logic/queries/account_holder_queries.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
@obs.api_call()
def get_account_holder(
    self,
    session: Session,
    /,
    id: AccountHolderId,
) -> AccountHolder:
    """
    Get an account holder entity from its ID.
    """
    with raise_if_account_holder_not_found(id):
        account_holder = AccountHolderModelBroker.get_account_holder(session, id=id)

    raise_on_terminated_account_holder(account_holder)

    return AccountHolder(
        id=AccountHolderId(account_holder.id),
        workspace_key=account_holder.workspace_key,
        external_id=account_holder.external_id,
        description=account_holder.description,
        reference=account_holder.reference,
    )

get_account_holder_id_by_external_id 

get_account_holder_id_by_external_id(
    session, /, workspace_key, external_id
)

Get an account holder entity from its external ID.

Source code in components/payment_gateway/subcomponents/accounts/protected/business_logic/queries/account_holder_queries.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@obs.api_call()
def get_account_holder_id_by_external_id(
    self,
    session: Session,
    /,
    workspace_key: str,
    external_id: str,
) -> AccountHolderId:
    """
    Get an account holder entity from its external ID.
    """
    with raise_if_account_holder_not_found_for_external_id(external_id):
        account_holder_id = (
            AccountHolderModelBroker.get_account_holder_id_by_external_id(
                session,
                workspace_key=workspace_key,
                external_id=external_id,
            )
        )

    return AccountHolderId(account_holder_id)

components.payment_gateway.subcomponents.accounts.protected.business_logic.queries.account_queries

AccountQueries

This class contains all the queries related to accounts.

get_account 

get_account(session, /, id)

Get an account entity from its ID.

Source code in components/payment_gateway/subcomponents/accounts/protected/business_logic/queries/account_queries.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
@obs.api_call()
def get_account(
    self,
    session: Session,
    /,
    id: AccountId,
) -> Account:
    """
    Get an account entity from its ID.
    """
    with raise_if_account_not_found(id):
        account = AccountModelBroker.get_account(session, id=id)

    raise_on_terminated_account(account)

    return _to_account(account)

get_account_id_by_external_id 

get_account_id_by_external_id(
    session, /, workspace_key, external_id
)

Get an account ID by its external ID.

Source code in components/payment_gateway/subcomponents/accounts/protected/business_logic/queries/account_queries.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
@obs.api_call()
def get_account_id_by_external_id(
    self,
    session: Session,
    /,
    workspace_key: str,
    external_id: str,
) -> AccountId:
    """
    Get an account ID by its external ID.
    """
    account_id = AccountModelBroker.get_account_id_by_external_id(
        session,
        workspace_key=workspace_key,
        external_id=external_id,
    )

    return AccountId(account_id)

get_account_ids_by_workspace 

get_account_ids_by_workspace(session, /, workspace_key)

Get all active account IDs for a workspace.

Source code in components/payment_gateway/subcomponents/accounts/protected/business_logic/queries/account_queries.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
@obs.api_call()
def get_account_ids_by_workspace(
    self,
    session: Session,
    /,
    workspace_key: str,
) -> list[AccountId]:
    """Get all active account IDs for a workspace."""
    return [
        AccountId(account.id)
        for account in AccountModelBroker.list_accounts_by_workspace(
            session, workspace_key=workspace_key
        )
    ]

get_account_ids_for_account_holder 

get_account_ids_for_account_holder(
    session, /, account_holder_id
)

Get all the account IDs for an account holder.

Source code in components/payment_gateway/subcomponents/accounts/protected/business_logic/queries/account_queries.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@obs.api_call()
def get_account_ids_for_account_holder(
    self,
    session: Session,
    /,
    account_holder_id: AccountHolderId,
) -> list[AccountId]:
    """
    Get all the account IDs for an account holder.
    """
    return [
        AccountId(account_id)
        for account_id in AccountModelBroker.list_account_ids_for_account_holder(
            session, account_holder_id=account_holder_id
        )
    ]

get_account_including_terminated 

get_account_including_terminated(session, /, id)

Get an account entity from its ID, including terminated ones.

Use this for read-only contexts (eg admin/marmot pages) where seeing a terminated account's data is a legitimate use case.

Source code in components/payment_gateway/subcomponents/accounts/protected/business_logic/queries/account_queries.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
@obs.api_call()
def get_account_including_terminated(
    self,
    session: Session,
    /,
    id: AccountId,
) -> Account:
    """
    Get an account entity from its ID, including terminated ones.

    Use this for read-only contexts (eg admin/marmot pages) where seeing
    a terminated account's data is a legitimate use case.
    """
    with raise_if_account_not_found(id):
        account = AccountModelBroker.get_account(session, id=id)

    return _to_account(account)

get_accounts_for_account_holder 

get_accounts_for_account_holder(
    session, /, account_holder_id
)

Get all the accounts for an account holder.

Source code in components/payment_gateway/subcomponents/accounts/protected/business_logic/queries/account_queries.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
@obs.api_call()
def get_accounts_for_account_holder(
    self,
    session: Session,
    /,
    account_holder_id: AccountHolderId,
) -> list[Account]:
    """
    Get all the accounts for an account holder.
    """
    return [
        _to_account(account)
        for account in AccountModelBroker.list_accounts_for_account_holder(
            session, account_holder_id=account_holder_id
        )
    ]

components.payment_gateway.subcomponents.accounts.protected.business_logic.queries.provider_account_queries

ProviderAccountQueries 

ProviderAccountQueries(adapter_registry)

Public queries for fetching a single provider account, with its live balances.

Source code in components/payment_gateway/subcomponents/accounts/protected/business_logic/queries/provider_account_queries.py 
47
48
def __init__(self, adapter_registry: ProviderAccountAdapterRegistry) -> None:
    self.adapter_registry = adapter_registry

adapter_registry instance-attribute 

adapter_registry = adapter_registry

create classmethod 

create()

Normal factory.

Source code in components/payment_gateway/subcomponents/accounts/protected/business_logic/queries/provider_account_queries.py 
27
28
29
30
@classmethod
def create(cls) -> "ProviderAccountQueries":
    """Normal factory."""
    return cls(adapter_registry=ProviderAccountAdapterRegistry.create())

create_null classmethod 

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

Null factory for tests.

Source code in components/payment_gateway/subcomponents/accounts/protected/business_logic/queries/provider_account_queries.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
@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]
) -> "ProviderAccountQueries":
    """Null factory for tests."""
    return cls(
        adapter_registry=ProviderAccountAdapterRegistry.create_null(
            track_requests=track_requests,
            responses=responses,
        )
    )

get_account 

get_account(*, account_id)

Return the provider account, with its live balances, backing account_id.

Resolves the PSP workspace and the account's external id from the local Account (terminated accounts included, so a closed Flex company account still exposes its balance), then fetches the account live from the PSP.

Source code in components/payment_gateway/subcomponents/accounts/protected/business_logic/queries/provider_account_queries.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
@obs.api_call()
def get_account(self, *, account_id: AccountId) -> ProviderAccount:
    """Return the provider account, with its live balances, backing ``account_id``.

    Resolves the PSP workspace and the account's external id from the local
    ``Account`` (terminated accounts included, so a closed Flex company
    account still exposes its balance), then fetches the account live from
    the PSP.
    """
    with raise_if_account_not_found(account_id):
        account = AccountModelBroker.get_account(current_session, id=account_id)

    provider = get_provider_for_workspace(account.workspace_key)
    if provider is None:
        raise WorkspaceNotRegisteredException(account.workspace_key)

    adapter = self.adapter_registry.get_account_adapter(provider)
    return adapter.get_account(external_id=account.external_id)