Skip to content

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 
11
12
13
14
15
16
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

ExtentedWebhookLogQueryResult module-attribute 

ExtentedWebhookLogQueryResult = tuple[
    WebhookLog,
    UUID | None,
    UUID | None,
    UUID | None,
    UUID | None,
    UUID | None,
    WebhookLogTransferStatus,
]

WebhookLogModelBroker

Bases: BaseModelBroker

get_webhook_log classmethod 
get_webhook_log(session, /, id)
Source code in components/payment_gateway/internal/models/brokers/webhook_log.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@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_extended_webhook_logs_in_period classmethod 
list_extended_webhook_logs_in_period(
    session,
    /,
    *,
    start_date,
    end_date,
    account_id=None,
    card_holder_id=None,
    status_filter=None,
)

Return a list of webhook logs with extended information about the transfer.

Args: - start_date: Start date to fetch event from (inclusive) - end_date: End date to fetch event from (inclusive) - account_id: Filter by account id - card_holder_id: Filter by card holder id - status_filter: Filter transfers based on their processing status. Can be: - WebhookLogTransferStatus.received: Webhook log status is not "ok" - WebhookLogTransferStatus.interrupted: Webhook log status is "ok" but internal id is not found - WebhookLogTransferStatus.processed: Means transfer was processed successfully

  • Tuple of
    • WebhookLog model
    • Payment id
    • Bank transfer id
    • Account transfer id
    • Account id
    • Card holder id
    • Transfer status
Source code in components/payment_gateway/internal/models/brokers/webhook_log.py 
 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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
@classmethod
def list_extended_webhook_logs_in_period(
    cls,
    session: Session,
    /,
    *,
    start_date: date,
    end_date: date,
    account_id: UUID | None = None,
    card_holder_id: UUID | None = None,
    status_filter: WebhookLogTransferStatus | None = None,
) -> list[ExtentedWebhookLogQueryResult]:
    """
    Return a list of webhook logs with extended information about the transfer.

    Args:
    - start_date: Start date to fetch event from (inclusive)
    - end_date: End date to fetch event from (inclusive)
    - account_id: Filter by account id
    - card_holder_id: Filter by card holder id
    - status_filter: Filter transfers based on their processing status. Can be:
        - WebhookLogTransferStatus.received: Webhook log status is not "ok"
        - WebhookLogTransferStatus.interrupted: Webhook log status is "ok" but internal id is not found
        - WebhookLogTransferStatus.processed: Means transfer was processed successfully

    Returns:
    - Tuple of
        - WebhookLog model
        - Payment id
        - Bank transfer id
        - Account transfer id
        - Account id
        - Card holder id
        - Transfer status

    """
    transfer_status = case(
        (WebhookLog.status != "ok", WebhookLogTransferStatus.received),
        (
            func.jsonb_extract_path_text(WebhookLog.payload, "data", "type")
            == "invoiceDeduction",
            WebhookLogTransferStatus.received,
        ),
        (
            and_(
                WebhookLog.status == "ok",
                CardTransfer.id.is_(None),
                BankTransfer.id.is_(None),
                AccountTransfer.id.is_(None),
            ),
            WebhookLogTransferStatus.ingested,
        ),
        else_=WebhookLogTransferStatus.routed,
    )

    query = (
        cls.select()
        .add_columns(
            CardTransfer.id.label("payment_id"),
            BankTransfer.id.label("bank_transfer_id"),
            AccountTransfer.id.label("account_transfer_id"),
            Account.id.label("account_id"),
            Card.card_holder_id.label("card_holder_id"),
            transfer_status.label("transfer_status"),
        )
        .outerjoin(
            CardTransfer,
            WebhookLog.transfer_external_id == CardTransfer.external_id,  # type: ignore[arg-type]
        )
        .outerjoin(
            BankTransfer,
            WebhookLog.transfer_external_id == BankTransfer.external_id,  # type: ignore[arg-type]
        )
        .outerjoin(
            AccountTransfer,
            WebhookLog.transfer_external_id == AccountTransfer.external_id,  # type: ignore[arg-type]
        )
        .outerjoin(Account, WebhookLog.account_external_id == Account.external_id)  # type: ignore[arg-type]
        .outerjoin(
            Card,
            WebhookLog.payment_instrument_external_id == Card.external_id,  # type: ignore[arg-type]
        )
        .filter(
            and_(
                WebhookLog.created_at.between(start_date, end_date),
                WebhookLog.is_transfer,
            )
        )
    )

    if account_id:
        query = query.filter(Account.id == account_id)
    if card_holder_id:
        query = query.filter(Card.card_holder_id == card_holder_id)
    if status_filter:
        query = query.filter(transfer_status == status_filter)

    webhook_logs = (
        session.execute(
            query.order_by(
                WebhookLog.transfer_external_id, desc(WebhookLog.sequence_number)
            ).distinct(WebhookLog.transfer_external_id)
        )
        .scalars()
        .all()
    )
    return list(webhook_logs)
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 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
@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.webhook_log

WebhookLog

Bases: BaseModel

Used to log incoming webhooks from Adyen.

We expect both issuing and relayed authorization webhooks from Adyen:

__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 
106
107
108
109
110
111
112
113
114
115
116
117
@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_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 
148
149
150
151
152
153
154
155
156
157
@amount.expression  # type: ignore[no-redef]
def amount(cls):
    """The Adyen amount for balance platform transfer events."""
    return case(
        (
            cls.is_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 
195
196
197
198
199
200
201
202
203
204
@events.expression  # type: ignore[no-redef]
def events(cls):
    """Return all events of a specific webhook log."""
    return case(
        (
            cls.is_transfer,
            func.jsonb_extract_path_text(cls.payload, "data", "events"),
        ),
        else_=None,
    )

is_transfer 

is_transfer()
Source code in components/payment_gateway/internal/models/webhook_log.py 
67
68
69
@is_transfer.expression  # type: ignore[no-redef]
def is_transfer(cls):
    return func.starts_with(cls.payload_type, "balancePlatform.transfer.")

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 
52
53
54
@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 
167
168
169
170
171
172
@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 
127
128
129
130
131
132
133
134
135
136
137
138
@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_transfer,
            func.jsonb_extract_path_text(
                cls.payload, "data", "paymentInstrument", "id"
            ),
        ),
        else_=None,
    )

sequence_number 

sequence_number()
Source code in components/payment_gateway/internal/models/webhook_log.py 
88
89
90
91
92
93
94
95
96
@sequence_number.expression  # type: ignore[no-redef]
def sequence_number(cls):
    return case(
        (
            cls.is_transfer,
            func.jsonb_extract_path(cls.payload, "data", "sequenceNumber"),
        ),
        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 
214
215
216
217
218
219
220
221
222
223
224
225
@transaction_rules.expression  # type: ignore[no-redef]
def transaction_rules(cls):
    """Return all transaction rules of a specific webhook log"""
    return case(
        (
            cls.is_transfer,
            func.jsonb_extract_path_text(
                cls.payload, "data", "transactionRulesResult"
            ),
        ),
        else_=None,
    )

transfer_external_id 

transfer_external_id()

The Adyen transfer ID for balance platform transfer events.

Source code in components/payment_gateway/internal/models/webhook_log.py 
75
76
77
78
79
80
81
@transfer_external_id.expression  # type: ignore[no-redef]
def transfer_external_id(cls):
    """The Adyen transfer ID for balance platform transfer events."""
    return case(
        (cls.is_transfer, 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 
178
179
180
181
182
183
184
185
186
187
@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_transfer,
            func.jsonb_extract_path_text(cls.payload, "data", "type"),
        ),
        else_=None,
    )