Skip to content

Business logic

components.payment_gateway.internal.business_logic.helpers

raise_on_inconsistent_workspace_keys 

raise_on_inconsistent_workspace_keys(*workspace_keys)

Helper to raise an InconsistentWorkspaceKeysException when the workspace keys are inconsistent

Source code in components/payment_gateway/internal/business_logic/helpers.py 
39
40
41
42
43
44
45
def raise_on_inconsistent_workspace_keys(
    *workspace_keys: str,
) -> None:
    """Helper to raise an InconsistentWorkspaceKeysException when the workspace keys are inconsistent"""
    unique_workspace_keys = set(workspace_keys)
    if len(unique_workspace_keys) != 1:
        raise InconsistentWorkspaceKeysException(list(unique_workspace_keys))

raise_on_provider_not_in 

raise_on_provider_not_in(workspace_key, allowed_providers)

Same as raise_on_provider_not_supported but allows any of a set of providers.

Source code in components/payment_gateway/internal/business_logic/helpers.py 
27
28
29
30
31
32
33
34
35
36
def raise_on_provider_not_in(
    workspace_key: str,
    allowed_providers: Iterable[PaymentServiceProvider],
) -> None:
    """Same as `raise_on_provider_not_supported` but allows any of a set of providers."""
    provider = get_provider_for_workspace(workspace_key)
    if provider is None:
        raise WorkspaceNotRegisteredException(workspace_key)
    if provider not in set(allowed_providers):
        raise ProviderNotSupportedException(provider)

raise_on_provider_not_supported 

raise_on_provider_not_supported(
    workspace_key, required_provider
)

Helper to raise: - ProviderNotSupportedException when the provider is not supported - WorkspaceNotRegisteredException when the workspace is not registered

Source code in components/payment_gateway/internal/business_logic/helpers.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
def raise_on_provider_not_supported(
    workspace_key: str,
    required_provider: PaymentServiceProvider,
) -> None:
    """Helper to raise:
    - ProviderNotSupportedException when the provider is not supported
    - WorkspaceNotRegisteredException when the workspace is not registered
    """
    provider = get_provider_for_workspace(workspace_key)
    if provider is None:
        raise WorkspaceNotRegisteredException(workspace_key)
    if provider != required_provider:
        raise ProviderNotSupportedException(provider)

components.payment_gateway.internal.business_logic.queries

psp_account_balances

PspAccountBalancesResult dataclass 

PspAccountBalancesResult(balances, failed_workspaces)

Outcome of fetching PSP balances, split between successes and failures.

balances instance-attribute 
balances

(workspace, balance) pairs successfully fetched.

failed_workspaces instance-attribute 
failed_workspaces

Workspaces whose PSP fetch raised an error (e.g. HTTP 500).

get_psp_account_balances 

get_psp_account_balances(workspaces)

Return account balances across the given PSP workspaces, tagged by workspace.

Resilient to per-PSP failures: an error fetching one workspace (e.g. a PSP HTTP 500) records that workspace as failed instead of breaking the whole call.

Args: - workspaces: Workspace keys to fetch balances for.

  • A PspAccountBalancesResult holding the successful (workspace, balance) pairs and the workspaces that failed. Workspaces whose provider is not registered are silently skipped (config, not a failure).
Source code in components/payment_gateway/internal/business_logic/queries/psp_account_balances.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def get_psp_account_balances(workspaces: list[str]) -> PspAccountBalancesResult:
    """Return account balances across the given PSP workspaces, tagged by workspace.

    Resilient to per-PSP failures: an error fetching one workspace (e.g. a PSP
    HTTP 500) records that workspace as failed instead of breaking the whole call.

    Args:
    - workspaces: Workspace keys to fetch balances for.

    Returns:
    - A ``PspAccountBalancesResult`` holding the successful ``(workspace, balance)``
      pairs and the workspaces that failed. Workspaces whose provider is not
      registered are silently skipped (config, not a failure).
    """
    queries = AccountBalanceQueries.create()

    balances: list[tuple[str, AccountBalance]] = []
    failed_workspaces: list[str] = []
    for workspace in workspaces:
        try:
            workspace_balances = queries.list_account_balances(workspace_key=workspace)
        except (ProviderNotSupportedException, WorkspaceNotRegisteredException):
            # Provider not implemented yet, skip this workspace
            continue
        except Exception:
            # External PSP boundary: surface the failure instead of breaking the
            # whole list so other PSPs still return their balances.
            current_logger.exception(
                "Failed to fetch PSP account balances", workspace=workspace
            )
            failed_workspaces.append(workspace)
            continue
        balances.extend((workspace, balance) for balance in workspace_balances)

    return PspAccountBalancesResult(
        balances=balances, failed_workspaces=failed_workspaces
    )

webhook_log_queries

get_latest_webhook_log_by_external_id 

get_latest_webhook_log_by_external_id(
    session, /, external_id
)

Get webhook log by id, a webhook log is a record of a webhook payload sent to the payment gateway from Adyen

Args: - webhook_id: The internal id of the webhook log to fetch

  • WebhookLog: The webhook log record
  • id: The internal id of the webhook log
  • payload: The raw content of the webhook payload from Adyen
Source code in components/payment_gateway/internal/business_logic/queries/webhook_log_queries.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def get_latest_webhook_log_by_external_id(
    session: Session,
    /,
    external_id: str,
) -> WebhookLogDataclass | None:
    """
    Get webhook log by id, a webhook log is a record of a webhook payload sent to the payment gateway from Adyen

    Args:
    - webhook_id: The internal id of the webhook log to fetch

    Returns:
    - WebhookLog: The webhook log record
      - id: The internal id of the webhook log
      - payload: The raw content of the webhook payload from Adyen
    """
    from components.payment_gateway.internal.models.brokers.webhook_log import (
        WebhookLogModelBroker,
    )

    webhook_logs = WebhookLogModelBroker.list_webhook_logs_by_external_id(
        session, external_id
    )
    sorted_webhook_logs = sorted(
        webhook_logs,
        key=lambda x: (x.sequence_number or -1, x.created_at),
        reverse=True,
    )
    if len(sorted_webhook_logs) == 0:
        return None

    return _to_dataclass(sorted_webhook_logs[0])

get_webhook_log 

get_webhook_log(session, /, webhook_id)

Get webhook log by id, a webhook log is a record of a webhook payload sent to the payment gateway from Adyen

Args: - webhook_id: The internal id of the webhook log to fetch

  • WebhookLog: The webhook log record
  • id: The internal id of the webhook log
  • payload: The raw content of the webhook payload from Adyen
Source code in components/payment_gateway/internal/business_logic/queries/webhook_log_queries.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def get_webhook_log(
    session: Session,
    /,
    webhook_id: UUID,
) -> WebhookLogDataclass:
    """
    Get webhook log by id, a webhook log is a record of a webhook payload sent to the payment gateway from Adyen

    Args:
    - webhook_id: The internal id of the webhook log to fetch

    Returns:
    - WebhookLog: The webhook log record
      - id: The internal id of the webhook log
      - payload: The raw content of the webhook payload from Adyen
    """
    from components.payment_gateway.internal.models.brokers.webhook_log import (
        WebhookLogModelBroker,
    )

    webhook_log = WebhookLogModelBroker.get_webhook_log(session, webhook_id)

    return _to_dataclass(webhook_log)