Models

components.payment_gateway.internal.models.base_model

BaseModel

Bases: BaseModel

Base class for all models in the Payment Gateway component.

__abstract__ class-attribute instance-attribute

__abstract__ = True

get_metabase_history_question_number

get_metabase_history_question_number()

Returns the Metabase history question number for this model.

This ensures that model histories are available from all backends.

Source code in components/payment_gateway/internal/models/base_model.py

def get_metabase_history_question_number(self) -> str | None:
    """Returns the Metabase history question number for this model.

    This ensures that model histories are available from all backends.
    """
    return current_config.get("METABASE_HISTORY_QUESTION_NUMBER_GLOBAL_COMPONENT")

components.payment_gateway.internal.models.brokers

webhook_log

WebhookLogModelBroker

Bases: BaseModelBroker

get_webhook_log classmethod

get_webhook_log(session, /, id)

Source code in components/payment_gateway/internal/models/brokers/webhook_log.py

@classmethod
def get_webhook_log(
    cls,
    session: Session,
    /,
    id: UUID,
) -> WebhookLog:
    webhook_log: WebhookLog = session.execute(
        cls.select().filter(WebhookLog.id == id)
    ).scalar_one()
    if webhook_log is None:
        raise BaseErrorCode.missing_resource(
            message=f"Webhook log not found for id: {id}"
        )
    return webhook_log

list_webhook_logs_by_external_id classmethod

list_webhook_logs_by_external_id(session, /, external_id)

Get webhook logs by transfer external ID.

Parameters:

Name	Type	Description	Default
external_id	str	The Adyen transfer ID to search for	required

Returns:

Type	Description
list[WebhookLog]	list[WebhookLog]: List of webhook log records

Source code in components/payment_gateway/internal/models/brokers/webhook_log.py

@classmethod
def list_webhook_logs_by_external_id(
    cls,
    session: Session,
    /,
    external_id: str,
) -> list[WebhookLog]:
    """Get webhook logs by transfer external ID.

    Args:
        external_id: The Adyen transfer ID to search for

    Returns:
        list[WebhookLog]: List of webhook log records
    """
    webhook_logs = (
        session.execute(
            cls.select().filter(cls.model.transfer_external_id == external_id)  # type: ignore[arg-type,comparison-overlap]
        )
        .scalars()
        .all()
    )
    if not webhook_logs:
        return []
    return list(webhook_logs)

model class-attribute instance-attribute

model = WebhookLog

components.payment_gateway.internal.models.helpers

PAYMENT_GATEWAY_SCHEMA_NAME module-attribute

PAYMENT_GATEWAY_SCHEMA_NAME = 'payment_gateway'

components.payment_gateway.internal.models.provider_entity_mixin

ProviderEntityMixin

Mixin for models that need to track workspace and external provider identifiers. Provides workspace_key and external_id columns with a unique constraint.

__table_args__

__table_args__()

Source code in components/payment_gateway/internal/models/provider_entity_mixin.py

@declared_attr.directive
def __table_args__(cls) -> tuple[Any, ...]:
    return cls._provider_entity_mixin_table_args()

external_id class-attribute instance-attribute

external_id = mapped_column(
    String(255), nullable=False, index=True
)

ID used by the external payment service provider to identify the entity in the workspace.

workspace_key class-attribute instance-attribute

workspace_key = mapped_column(String(50), nullable=False)

Key used to identify the PSP workspace.

components.payment_gateway.internal.models.webhook_log

WebhookLog

Bases: BaseModel

Used to log incoming webhooks from payment providers (Adyen, Revolut ...).

We expect both issuing and relayed authorization webhooks from Adyen:

• Issuing webhooks are used to notify us of events on the balance platform. See https://docs.adyen.com/issuing/webhook-types ⧉ for more information about the expected payloads.
• Relayed authorization webhooks use a different format that is currently undocumented; recording their payload there will help us understand how to implement this feature. See https://docs.adyen.com/issuing/authorisation/relayed-authorisation/ ⧉ for more information.

We expect transfer Revolut webhooks: - TransactionCreated: Notifies of new transactions - TransactionStateChanged: Notifies of transaction state updates See https://developer.revolut.com/docs/business/webhooks-v-2 ⧉

__table_args__ class-attribute instance-attribute

__table_args__ = {'schema': PAYMENT_GATEWAY_SCHEMA_NAME}

__tablename__ class-attribute instance-attribute

__tablename__ = 'webhook_log'

account_external_id

account_external_id()

The Adyen account ID for balance platform transfer events.

Source code in components/payment_gateway/internal/models/webhook_log.py

@account_external_id.expression  # type: ignore[no-redef]
def account_external_id(cls):
    """The Adyen account ID for balance platform transfer events."""
    return case(
        (
            cls.is_adyen_transfer,
            func.jsonb_extract_path_text(
                cls.payload, "data", "balanceAccount", "id"
            ),
        ),
        else_=None,
    )

amount

amount()

The Adyen amount for balance platform transfer events.

Source code in components/payment_gateway/internal/models/webhook_log.py

@amount.expression  # type: ignore[no-redef]
def amount(cls):
    """The Adyen amount for balance platform transfer events."""
    return case(
        (
            cls.is_adyen_transfer,
            func.jsonb_extract_path_text(cls.payload, "data", "amount", "value"),
        ),
        else_=None,
    )

events

events()

Return all events of a specific webhook log.

Source code in components/payment_gateway/internal/models/webhook_log.py

@events.expression  # type: ignore[no-redef]
def events(cls):
    """Return all events of a specific webhook log."""
    return case(
        (
            cls.is_adyen_transfer,
            func.jsonb_extract_path_text(cls.payload, "data", "events"),
        ),
        else_=None,
    )

is_adyen_transfer

is_adyen_transfer()

Source code in components/payment_gateway/internal/models/webhook_log.py

@is_adyen_transfer.expression  # type: ignore[no-redef]
def is_adyen_transfer(cls):
    return func.starts_with(cls.payload_type, "balancePlatform.transfer.")

is_revolut

is_revolut()

Source code in components/payment_gateway/internal/models/webhook_log.py

@is_revolut.expression  # type: ignore[no-redef]
def is_revolut(cls):
    return func.lower(cls.source).contains("revolut")

payload class-attribute instance-attribute

payload = mapped_column(
    JSONB(none_as_null=True), nullable=False
)

Raw JSON payload of the webhook.

payload_type

payload_type()

Source code in components/payment_gateway/internal/models/webhook_log.py

@payload_type.expression  # type: ignore[no-redef]
def payload_type(cls):
    return func.jsonb_extract_path_text(cls.payload, "type")

payment_date

payment_date()

The booking date of the received event for balance platform transfer events.

Source code in components/payment_gateway/internal/models/webhook_log.py

@payment_date.expression  # type: ignore[no-redef]
def payment_date(cls):
    """The booking date of the received event for balance platform transfer events."""
    return func.jsonb_path_query_first(
        cls.payload, '$.data.events[*] ? (@.status == "received").bookingDate'
    )

payment_instrument_external_id

payment_instrument_external_id()

The Adyen payment instrument ID for balance platform transfer events.

Source code in components/payment_gateway/internal/models/webhook_log.py

@payment_instrument_external_id.expression  # type: ignore[no-redef]
def payment_instrument_external_id(cls):
    """The Adyen payment instrument ID for balance platform transfer events."""
    return case(
        (
            cls.is_adyen_transfer,
            func.jsonb_extract_path_text(
                cls.payload, "data", "paymentInstrument", "id"
            ),
        ),
        else_=None,
    )

revolut_event_type

revolut_event_type()

Source code in components/payment_gateway/internal/models/webhook_log.py

@revolut_event_type.expression  # type: ignore[no-redef]
def revolut_event_type(cls):
    return case(
        (cls.is_revolut, func.jsonb_extract_path_text(cls.payload, "event")),
        else_=None,
    )

sequence_number

sequence_number()

Source code in components/payment_gateway/internal/models/webhook_log.py

@sequence_number.expression  # type: ignore[no-redef]
def sequence_number(cls):
    return case(
        (
            cls.is_adyen_transfer,
            func.jsonb_extract_path(cls.payload, "data", "sequenceNumber"),
        ),
        (
            cls.is_revolut,
            case(
                (
                    func.jsonb_extract_path_text(cls.payload, "event")
                    == "TransactionCreated",
                    1,
                ),
                (
                    func.jsonb_extract_path_text(cls.payload, "event")
                    == "TransactionStateChanged",
                    2,
                ),
                else_=None,
            ),
        ),
        else_=None,
    )

source class-attribute instance-attribute

source = mapped_column(String(255), nullable=False)

The source of the webhook, usually the name of the controller receiving the webhook.

status class-attribute instance-attribute

status = mapped_column(
    AlanBaseEnumTypeDecorator(WebhookStatus), nullable=False
)

Whether the webhook was successfully processed or not at the controller level.

transaction_rules

transaction_rules()

Return all transaction rules of a specific webhook log

Source code in components/payment_gateway/internal/models/webhook_log.py

@transaction_rules.expression  # type: ignore[no-redef]
def transaction_rules(cls):
    """Return all transaction rules of a specific webhook log"""
    return case(
        (
            cls.is_adyen_transfer,
            func.jsonb_extract_path_text(
                cls.payload, "data", "transactionRulesResult"
            ),
        ),
        else_=None,
    )

transfer_external_id

transfer_external_id()

Source code in components/payment_gateway/internal/models/webhook_log.py

@transfer_external_id.expression  # type: ignore[no-redef]
def transfer_external_id(cls):
    return case(
        (
            cls.is_adyen_transfer,
            func.jsonb_extract_path_text(cls.payload, "data", "id"),
        ),
        (cls.is_revolut, func.jsonb_extract_path_text(cls.payload, "data", "id")),
        else_=None,
    )

transfer_type

transfer_type()

Transfer type of this Adyen webhook between internalTransfer, bankTransfer and payment

Source code in components/payment_gateway/internal/models/webhook_log.py

@transfer_type.expression  # type: ignore[no-redef]
def transfer_type(cls):
    """Transfer type of this Adyen webhook between internalTransfer, bankTransfer and payment"""
    return case(
        (
            cls.is_adyen_transfer,
            func.jsonb_extract_path_text(cls.payload, "data", "type"),
        ),
        else_=None,
    )