Model brokers

components.payment_gateway.subcomponents.payments.models.brokers.payment_request

PaymentRequestModelBroker

Bases: BaseModelBroker

create_payment_request classmethod

create_payment_request(
    session,
    /,
    *,
    workspace_key,
    external_id,
    account_id,
    payment_type,
    reference,
    amount,
    currency,
    raw=None,
    bank_transfer_id=None,
    financial_instrument_id=None,
)

Source code in components/payment_gateway/subcomponents/payments/models/brokers/payment_request.py

@classmethod
def create_payment_request(
    cls,
    session: Session,
    /,
    *,
    workspace_key: str,
    external_id: str,
    account_id: UUID,
    payment_type: PaymentRequestType,
    reference: str,
    amount: int,
    currency: CurrencyCode,
    raw: dict[str, Any] | None = None,
    bank_transfer_id: UUID | None = None,
    financial_instrument_id: UUID | None = None,
) -> PaymentRequest:
    payment_request = cls.model(
        workspace_key=workspace_key,
        external_id=external_id,
        account_id=account_id,
        type=payment_type,
        reference=reference,
        amount=amount,
        currency=currency,
        raw=raw,
        bank_transfer_id=bank_transfer_id,
        financial_instrument_id=financial_instrument_id,
    )
    session.add(payment_request)
    session.flush()

    return payment_request

get_payment_request classmethod

get_payment_request(session, /, id)

Source code in components/payment_gateway/subcomponents/payments/models/brokers/payment_request.py

@classmethod
def get_payment_request(
    cls,
    session: Session,
    /,
    id: UUID,
) -> PaymentRequest:
    payment_request = session.execute(
        select(cls.model).filter(
            PaymentRequest.id == id,
        )
    ).scalar_one()
    return payment_request

get_payment_request_by_external_id classmethod

get_payment_request_by_external_id(
    session, /, *, workspace_key, external_id
)

Source code in components/payment_gateway/subcomponents/payments/models/brokers/payment_request.py

@classmethod
def get_payment_request_by_external_id(
    cls,
    session: Session,
    /,
    *,
    workspace_key: str,
    external_id: str,
) -> PaymentRequest:
    payment_request = session.execute(
        select(cls.model).filter(
            PaymentRequest.workspace_key == workspace_key,
            PaymentRequest.external_id == external_id,
        )
    ).scalar_one()
    return payment_request

get_stale_payment_requests classmethod

get_stale_payment_requests(
    session,
    /,
    *,
    workspace_keys,
    pending_status,
    staleness_threshold,
    limit,
    offset=0,
    exclude_ids=None,
)

Return PaymentRequest ORM models with no linked BankTransfer or whose latest TransferUpdate status matches pending_status.

Source code in components/payment_gateway/subcomponents/payments/models/brokers/payment_request.py

@classmethod
def get_stale_payment_requests(
    cls,
    session: Session,
    /,
    *,
    workspace_keys: list[str],
    pending_status: str,
    staleness_threshold: timedelta,
    limit: int,
    offset: int = 0,
    exclude_ids: list[UUID] | None = None,
) -> list[PaymentRequest]:
    """Return PaymentRequest ORM models with no linked BankTransfer or whose latest
    TransferUpdate status matches `pending_status`."""
    from components.payment_gateway.subcomponents.transfers.models.transfer_update import (
        TransferUpdate as TransferUpdateModel,
    )

    latest_transfer_update_status = (
        select(TransferUpdateModel.status)
        .where(TransferUpdateModel.transfer_id == PaymentRequest.bank_transfer_id)
        .order_by(TransferUpdateModel.sequence_number.desc())
        .limit(1)
        .correlate(PaymentRequest)
        .scalar_subquery()
    )
    stmt = (
        select(cls.model)
        .where(
            PaymentRequest.workspace_key.in_(workspace_keys),
            PaymentRequest.created_at < datetime.now(UTC) - staleness_threshold,
            or_(
                PaymentRequest.bank_transfer_id.is_(None),
                latest_transfer_update_status == pending_status,
            ),
        )
        .order_by(PaymentRequest.created_at, PaymentRequest.id)
        .offset(offset)
        .limit(limit)
    )
    if exclude_ids:
        stmt = stmt.where(PaymentRequest.id.notin_(exclude_ids))
    return list(session.scalars(stmt).all())

model class-attribute instance-attribute

model = PaymentRequest

set_bank_transfer classmethod

set_bank_transfer(session, /, *, id, bank_transfer_id)

Set the bank transfer for a payment request.

Parameters:

Name	Type	Description	Default
session	Session	Database session	required
id	UUID	ID of the payment request	required
bank_transfer_id	UUID	ID of the bank transfer to link	required

Returns:

Type	Description
PaymentRequest	Updated PaymentRequest instance

Source code in components/payment_gateway/subcomponents/payments/models/brokers/payment_request.py

@classmethod
def set_bank_transfer(
    cls,
    session: Session,
    /,
    *,
    id: UUID,
    bank_transfer_id: UUID,
) -> PaymentRequest:
    """
    Set the bank transfer for a payment request.

    Args:
        session: Database session
        id: ID of the payment request
        bank_transfer_id: ID of the bank transfer to link

    Returns:
        Updated PaymentRequest instance
    """
    payment_request = cls.get_payment_request(session, id=id)

    payment_request.bank_transfer_id = bank_transfer_id

    return payment_request