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 
25
26
27
28
29
30
31
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_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 
10
11
12
13
14
15
16
17
18
19
20
21
22
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

webhook_log_extended_queries

get_extended_webhook_logs_in_period 

get_extended_webhook_logs_in_period(
    session,
    /,
    from_date,
    to_date,
    account_id=None,
    card_holder_id=None,
    status_filter=None,
)
Source code in components/payment_gateway/internal/business_logic/queries/webhook_log_extended_queries.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def get_extended_webhook_logs_in_period(
    session: Session,
    /,
    from_date: date,
    to_date: date,
    account_id: UUID | None = None,
    card_holder_id: UUID | None = None,
    status_filter: WebhookLogTransferStatus | None = None,
) -> list[WebhookLogExtended]:
    webhook_logs_in_period = WebhookLogModelBroker.list_extended_webhook_logs_in_period(
        session,
        start_date=from_date,
        end_date=to_date,
        account_id=account_id,
        card_holder_id=card_holder_id,
        status_filter=status_filter,
    )

    return [
        _webhook_log_extended_to_dataclass(webhook_log_extended_tuple)
        for webhook_log_extended_tuple in webhook_logs_in_period
    ]

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)