Model brokers

components.payment_gateway.subcomponents.transfers.models.brokers.account_transfer

AccountTransferModelBroker

Bases: BaseModelBroker

Model broker for the AccountTransfer model.

aggregate_account_transfer_balance classmethod

aggregate_account_transfer_balance(
    session,
    /,
    transfer_history_id,
    *,
    effective_date,
    creation_date,
)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/account_transfer.py

@classmethod
def aggregate_account_transfer_balance(
    cls,
    session: Session,
    /,
    transfer_history_id: UUID,
    *,
    effective_date: datetime,
    creation_date: datetime,
) -> BalanceInformation:
    stmt = (
        select(
            func.sum(TransferEvent.received),
            func.sum(TransferEvent.reserved),
            func.sum(TransferEvent.balance),
        )
        .join(AccountTransferTransferEventAssociation)
        .join(AccountTransfer)
        .where(AccountTransfer.transfer_history_id == transfer_history_id)
        .where(AccountTransfer.effective_date <= effective_date)
        .where(AccountTransfer.created_at <= creation_date)
        .where(TransferEvent.effective_date <= effective_date)
        .where(TransferEvent.created_at <= creation_date)
    )

    # this will a tuple of (None, None, None) if there are no results
    result = session.execute(stmt).one()

    return BalanceInformation(
        received=result[0] or 0,
        reserved=result[1] or 0,
        balance=result[2] or 0,
    )

autoload class-attribute instance-attribute

autoload = {'events': True, 'updates': True}

find_account_transfer_by_id classmethod

find_account_transfer_by_id(session, /, id)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/account_transfer.py

@classmethod
def find_account_transfer_by_id(
    cls,
    session: Session,
    /,
    id: UUID,
) -> AccountTransfer | None:
    return session.execute(
        cls.select().filter(AccountTransfer.id == id)
    ).scalar_one_or_none()

get_account_transfer classmethod

get_account_transfer(session, /, id)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/account_transfer.py

@classmethod
def get_account_transfer(
    cls,
    session: Session,
    /,
    id: UUID,
) -> AccountTransfer:
    account_transfer: AccountTransfer = session.execute(
        cls.select().filter(AccountTransfer.id == id)
    ).scalar_one()
    return account_transfer

get_account_transfer_by_external_id classmethod

get_account_transfer_by_external_id(
    session, /, provider, external_id
)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/account_transfer.py

@classmethod
def get_account_transfer_by_external_id(
    cls,
    session: Session,
    /,
    provider: PaymentServiceProvider,
    external_id: str,
) -> AccountTransfer:
    account_transfer: AccountTransfer = session.execute(
        cls.select().filter(
            AccountTransfer.provider == provider,
            AccountTransfer.external_id == external_id,
        )
    ).scalar_one()
    return account_transfer

list_account_transfers_for_accounts classmethod

list_account_transfers_for_accounts(
    session,
    /,
    account_ids,
    *,
    start_effective_date=None,
    end_effective_date=None,
)

Find all the account transfers for the given accounts and optional time period.

Note: this will include TransferEvents that are out of bounds of the given time period. If at some point we want a function that returns only the events in the given time period, we should create a new method based on this ⧉ and not forget to add the .outerjoin.

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/account_transfer.py

@classmethod
def list_account_transfers_for_accounts(
    cls,
    session: Session,
    /,
    account_ids: list["AccountId"],
    *,
    start_effective_date: datetime | None = None,
    end_effective_date: datetime | None = None,
) -> list[AccountTransfer]:
    """Find all the account transfers for the given accounts and optional time period.

    Note: this will include TransferEvents that are out of bounds of the given time period.
    If at some point we want a function that returns only the events in the given time period,
    we should create a new method based on [this](https://stackoverflow.com/questions/67578639/flask-sqlalchemy-filter-objects-in-relationship-for-each-object)
    and not forget to add the .outerjoin.

    """
    return cls.search_account_transfers(
        session,
        account_ids=account_ids,
        start_effective_date=start_effective_date,
        end_effective_date=end_effective_date,
    )

list_account_transfers_for_transfer_histories classmethod

list_account_transfers_for_transfer_histories(
    session,
    /,
    transfer_history_ids,
    *,
    start_effective_date=None,
    end_effective_date=None,
    order_by_created_at=False,
)

Find all the account transfers for the given transfer histories and optional time period.

Note: this will include TransferEvents that are out of bounds of the given time period. If at some point we want a function that returns only the events in the given time period, we should create a new method based on this ⧉ and not forget to add the .outerjoin.

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/account_transfer.py

@classmethod
def list_account_transfers_for_transfer_histories(
    cls,
    session: Session,
    /,
    transfer_history_ids: list["TransferHistoryId"],
    *,
    start_effective_date: datetime | None = None,
    end_effective_date: datetime | None = None,
    order_by_created_at: bool = False,
) -> list[AccountTransfer]:
    """Find all the account transfers for the given transfer histories and optional time period.

    Note: this will include TransferEvents that are out of bounds of the given time period.
    If at some point we want a function that returns only the events in the given time period,
    we should create a new method based on [this](https://stackoverflow.com/questions/67578639/flask-sqlalchemy-filter-objects-in-relationship-for-each-object)
    and not forget to add the .outerjoin.

    """
    return cls.search_account_transfers(
        session,
        transfer_history_ids=transfer_history_ids,
        start_effective_date=start_effective_date,
        end_effective_date=end_effective_date,
        order_by_created_at=order_by_created_at,
    )

model class-attribute instance-attribute

model = AccountTransfer

record_account_transfer classmethod

record_account_transfer(
    session,
    /,
    *,
    effective_date,
    direction,
    reference,
    provider,
    external_id,
    account_id,
    transfer_history_id,
    raw=None,
)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/account_transfer.py

@classmethod
def record_account_transfer(
    cls,
    session: Session,
    /,
    *,
    effective_date: datetime,
    direction: TransferDirection,
    reference: str,
    provider: PaymentServiceProvider,
    external_id: str,
    account_id: UUID,
    transfer_history_id: UUID,
    raw: dict[str, Any] | None = None,
) -> tuple[AccountTransfer, bool]:
    insert_stmt = (
        insert(AccountTransfer)
        .values(
            effective_date=effective_date,
            direction=direction,
            reference=reference,
            provider=provider,
            external_id=external_id,
            account_id=account_id,
            transfer_history_id=transfer_history_id,
            raw=raw,
        )
        .on_conflict_do_nothing(
            constraint="account_transfer__unique_external_id_per_provider"
        )
        .returning(cls.model.id)
    )
    created = session.scalar(insert_stmt)

    # Return model from provider-specific IDs
    return session.execute(
        cls.select().filter(
            cls.model.provider == provider,
            cls.model.external_id == external_id,
        )
    ).scalar_one(), bool(created)

search_account_transfers classmethod

search_account_transfers(
    session,
    /,
    *,
    account_ids=None,
    transfer_history_ids=None,
    start_effective_date=None,
    end_effective_date=None,
    order_by_created_at=False,
)

Find all the account transfers for the given time period.

Note: this will include TransferEvents that are out of bounds of the given time period. If at some point we want a function that returns only the events in the given time period, we should create a new method based on this ⧉ and not forget to add the .outerjoin.

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/account_transfer.py

@classmethod
def search_account_transfers(
    cls,
    session: Session,
    /,
    *,
    account_ids: list["AccountId"] | None = None,
    transfer_history_ids: list["TransferHistoryId"] | None = None,
    start_effective_date: datetime | None = None,
    end_effective_date: datetime | None = None,
    order_by_created_at: bool = False,
) -> list[AccountTransfer]:
    """Find all the account transfers for the given time period.

    Note: this will include TransferEvents that are out of bounds of the given time period.
    If at some point we want a function that returns only the events in the given time period,
    we should create a new method based on [this](https://stackoverflow.com/questions/67578639/flask-sqlalchemy-filter-objects-in-relationship-for-each-object)
    and not forget to add the .outerjoin.

    """
    query = cls.select()
    if account_ids is not None:
        query = query.filter(AccountTransfer.account_id.in_(account_ids))
    if transfer_history_ids is not None:
        query = query.filter(
            AccountTransfer.transfer_history_id.in_(transfer_history_ids)
        )

    if start_effective_date:
        query = query.filter(AccountTransfer.effective_date >= start_effective_date)
    if end_effective_date:
        query = query.filter(AccountTransfer.effective_date <= end_effective_date)

    # Build ordering
    orderings = []
    if order_by_created_at:
        orderings.append(AccountTransfer.created_at.desc())
    orderings.append(AccountTransfer.effective_date.desc())

    query = query.order_by(*orderings)

    return list(session.execute(query).scalars().all())

components.payment_gateway.subcomponents.transfers.models.brokers.bank_transfer

BankTransferModelBroker

Bases: BaseModelBroker

Model broker for the BankTransfer model.

aggregate_bank_transfer_balance classmethod

aggregate_bank_transfer_balance(
    session,
    /,
    transfer_history_id,
    *,
    effective_date,
    creation_date,
)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/bank_transfer.py

@classmethod
def aggregate_bank_transfer_balance(
    cls,
    session: Session,
    /,
    transfer_history_id: UUID,
    *,
    effective_date: datetime,
    creation_date: datetime,
) -> BalanceInformation:
    stmt = (
        select(
            func.sum(TransferEvent.received),
            func.sum(TransferEvent.reserved),
            func.sum(TransferEvent.balance),
        )
        .join(BankTransferTransferEventAssociation)
        .join(BankTransfer)
        .where(BankTransfer.transfer_history_id == transfer_history_id)
        .where(BankTransfer.effective_date <= effective_date)
        .where(BankTransfer.created_at <= creation_date)
        .where(TransferEvent.effective_date <= effective_date)
        .where(TransferEvent.created_at <= creation_date)
    )

    # this will a tuple of (None, None, None) if there are no results
    result = session.execute(stmt).one()

    return BalanceInformation(
        received=result[0] or 0,
        reserved=result[1] or 0,
        balance=result[2] or 0,
    )

autoload class-attribute instance-attribute

autoload = {'events': True, 'updates': True}

find_bank_transfer_by_id classmethod

find_bank_transfer_by_id(session, /, id)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/bank_transfer.py

@classmethod
def find_bank_transfer_by_id(
    cls,
    session: Session,
    /,
    id: UUID,
) -> BankTransfer | None:
    return session.execute(
        cls.select().filter(BankTransfer.id == id)
    ).scalar_one_or_none()

get_bank_transfer classmethod

get_bank_transfer(session, /, id, with_sepa_mandate=False)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/bank_transfer.py

@classmethod
def get_bank_transfer(
    cls,
    session: Session,
    /,
    id: UUID,
    with_sepa_mandate: bool = False,
) -> BankTransfer:
    if with_sepa_mandate:
        query = cls.select(custom_autoload=cls.autoload | {"sepa_mandate": True})
    else:
        query = cls.select()
    bank_transfer: BankTransfer = session.execute(
        query.filter(BankTransfer.id == id)
    ).scalar_one()
    return bank_transfer

get_bank_transfer_by_external_id classmethod

get_bank_transfer_by_external_id(
    session, /, provider, external_id
)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/bank_transfer.py

@classmethod
def get_bank_transfer_by_external_id(
    cls,
    session: Session,
    /,
    provider: PaymentServiceProvider,
    external_id: str,
) -> BankTransfer:
    bank_transfer: BankTransfer = session.execute(
        cls.select().filter(
            BankTransfer.provider == provider,
            BankTransfer.external_id == external_id,
        )
    ).scalar_one()
    return bank_transfer

list_bank_transfers_for_accounts classmethod

list_bank_transfers_for_accounts(
    session,
    /,
    account_ids,
    *,
    start_effective_date=None,
    end_effective_date=None,
)

Find all the bank transfers for the given transfer histories and optional time period.

Note: this will include TransferEvents that are out of bounds of the given time period. If at some point we want a function that returns only the events in the given time period, we should create a new method based on this ⧉ and not forget to add the .outerjoin.

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/bank_transfer.py

@classmethod
def list_bank_transfers_for_accounts(
    cls,
    session: Session,
    /,
    account_ids: list["AccountId"],
    *,
    start_effective_date: datetime | None = None,
    end_effective_date: datetime | None = None,
) -> list[BankTransfer]:
    """Find all the bank transfers for the given transfer histories and optional time period.

    Note: this will include TransferEvents that are out of bounds of the given time period.
    If at some point we want a function that returns only the events in the given time period,
    we should create a new method based on [this](https://stackoverflow.com/questions/67578639/flask-sqlalchemy-filter-objects-in-relationship-for-each-object)
    and not forget to add the .outerjoin.

    """
    return cls.search_bank_transfers(
        session,
        account_ids=account_ids,
        start_effective_date=start_effective_date,
        end_effective_date=end_effective_date,
    )

list_bank_transfers_for_transfer_histories classmethod

list_bank_transfers_for_transfer_histories(
    session,
    /,
    transfer_history_ids,
    *,
    start_effective_date=None,
    end_effective_date=None,
    order_by_created_at=False,
)

Find all the bank transfers for the given transfer histories and optional time period.

Note: this will include TransferEvents that are out of bounds of the given time period. If at some point we want a function that returns only the events in the given time period, we should create a new method based on this ⧉ and not forget to add the .outerjoin.

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/bank_transfer.py

@classmethod
def list_bank_transfers_for_transfer_histories(
    cls,
    session: Session,
    /,
    transfer_history_ids: list["TransferHistoryId"],
    *,
    start_effective_date: datetime | None = None,
    end_effective_date: datetime | None = None,
    order_by_created_at: bool = False,
) -> list[BankTransfer]:
    """Find all the bank transfers for the given transfer histories and optional time period.

    Note: this will include TransferEvents that are out of bounds of the given time period.
    If at some point we want a function that returns only the events in the given time period,
    we should create a new method based on [this](https://stackoverflow.com/questions/67578639/flask-sqlalchemy-filter-objects-in-relationship-for-each-object)
    and not forget to add the .outerjoin.

    """
    return cls.search_bank_transfers(
        session,
        transfer_history_ids=transfer_history_ids,
        start_effective_date=start_effective_date,
        end_effective_date=end_effective_date,
        order_by_created_at=order_by_created_at,
    )

model class-attribute instance-attribute

model = BankTransfer

record_bank_transfer classmethod

record_bank_transfer(
    session,
    /,
    *,
    effective_date,
    direction,
    provider,
    external_id,
    account_id,
    transfer_history_id,
    sepa_mandate_id=None,
    sepa_beneficiary_id=None,
    raw=None,
)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/bank_transfer.py

@classmethod
def record_bank_transfer(
    cls,
    session: Session,
    /,
    *,
    effective_date: datetime,
    direction: TransferDirection,
    provider: PaymentServiceProvider,
    external_id: str,
    account_id: UUID,
    transfer_history_id: UUID,
    sepa_mandate_id: UUID | None = None,
    sepa_beneficiary_id: UUID | None = None,
    raw: dict[str, Any] | None = None,
) -> tuple[BankTransfer, bool]:
    insert_stmt = (
        insert(BankTransfer)
        .values(
            effective_date=effective_date,
            direction=direction,
            provider=provider,
            external_id=external_id,
            account_id=account_id,
            transfer_history_id=transfer_history_id,
            sepa_mandate_id=sepa_mandate_id,
            sepa_beneficiary_id=sepa_beneficiary_id,
            raw=raw,
        )
        .on_conflict_do_nothing(
            constraint="bank_transfer__unique_external_id_per_provider"
        )
        .returning(cls.model.id)
    )
    created = session.scalar(insert_stmt)

    # Return model from provider-specific IDs
    return session.execute(
        cls.select().filter(
            cls.model.provider == provider,
            cls.model.external_id == external_id,
        )
    ).scalar_one(), bool(created)

search_bank_transfers classmethod

search_bank_transfers(
    session,
    /,
    *,
    account_ids=None,
    transfer_history_ids=None,
    start_effective_date=None,
    end_effective_date=None,
    order_by_created_at=False,
)

Find all the bank transfers in a transfer history for the given time period.

Note: this will include TransferEvents that are out of bounds of the given time period. If at some point we want a function that returns only the events in the given time period, we should create a new method based on this ⧉ and not forget to add the .outerjoin.

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/bank_transfer.py

@classmethod
def search_bank_transfers(
    cls,
    session: Session,
    /,
    *,
    account_ids: list["AccountId"] | None = None,
    transfer_history_ids: list["TransferHistoryId"] | None = None,
    start_effective_date: datetime | None = None,
    end_effective_date: datetime | None = None,
    order_by_created_at: bool = False,
) -> list[BankTransfer]:
    """Find all the bank transfers in a transfer history for the given time period.

    Note: this will include TransferEvents that are out of bounds of the given time period.
    If at some point we want a function that returns only the events in the given time period,
    we should create a new method based on [this](https://stackoverflow.com/questions/67578639/flask-sqlalchemy-filter-objects-in-relationship-for-each-object)
    and not forget to add the .outerjoin.

    """
    query = cls.select()
    if account_ids is not None:
        query = query.filter(BankTransfer.account_id.in_(account_ids))
    if transfer_history_ids is not None:
        query = query.filter(
            BankTransfer.transfer_history_id.in_(transfer_history_ids)
        )

    if start_effective_date:
        query = query.filter(BankTransfer.effective_date >= start_effective_date)
    if end_effective_date:
        query = query.filter(BankTransfer.effective_date <= end_effective_date)

    # Build ordering
    orderings = []
    if order_by_created_at:
        orderings.append(BankTransfer.created_at.desc())
    orderings.append(BankTransfer.effective_date.desc())

    query = query.order_by(*orderings)

    return list(session.execute(query).scalars().all())

components.payment_gateway.subcomponents.transfers.models.brokers.card_transfer

CardTransferModelBroker

Bases: BaseModelBroker

Model broker for the CardTransfer model.

aggregate_card_transfer_balance classmethod

aggregate_card_transfer_balance(
    session,
    /,
    transfer_history_id,
    *,
    effective_date,
    creation_date,
)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/card_transfer.py

@classmethod
def aggregate_card_transfer_balance(
    cls,
    session: Session,
    /,
    transfer_history_id: UUID,
    *,
    effective_date: datetime,
    creation_date: datetime,
) -> BalanceInformation:
    stmt = (
        select(
            func.sum(TransferEvent.received),
            func.sum(TransferEvent.reserved),
            func.sum(TransferEvent.balance),
        )
        .join(CardTransferTransferEventAssociation)
        .join(CardTransfer)
        .where(CardTransfer.transfer_history_id == transfer_history_id)
        .where(CardTransfer.effective_date <= effective_date)
        .where(CardTransfer.created_at <= creation_date)
        .where(TransferEvent.effective_date <= effective_date)
        .where(TransferEvent.created_at <= creation_date)
    )

    # this will a tuple of (None, None, None) if there are no results
    result = session.execute(stmt).one()

    return BalanceInformation(
        received=result[0] or 0,
        reserved=result[1] or 0,
        balance=result[2] or 0,
    )

autoload class-attribute instance-attribute

autoload = {'updates': True, 'events': True}

find_card_transfer_by_id classmethod

find_card_transfer_by_id(session, /, id)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/card_transfer.py

@classmethod
def find_card_transfer_by_id(
    cls,
    session: Session,
    /,
    id: UUID,
) -> CardTransfer | None:
    return session.execute(
        cls.select().filter(CardTransfer.id == id)
    ).scalar_one_or_none()

get_card_transfer classmethod

get_card_transfer(session, /, id)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/card_transfer.py

@classmethod
def get_card_transfer(
    cls,
    session: Session,
    /,
    id: UUID,
) -> CardTransfer:
    card_transfer: CardTransfer = session.execute(
        cls.select().filter(CardTransfer.id == id)
    ).scalar_one()
    return card_transfer

get_card_transfer_by_external_id classmethod

get_card_transfer_by_external_id(
    session, /, provider, external_id
)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/card_transfer.py

@classmethod
def get_card_transfer_by_external_id(
    cls,
    session: Session,
    /,
    provider: PaymentServiceProvider,
    external_id: str,
) -> CardTransfer:
    card_transfer: CardTransfer = session.execute(
        cls.select().filter(
            CardTransfer.provider == provider,
            CardTransfer.external_id == external_id,
        )
    ).scalar_one()
    return card_transfer

list_card_transfers_for_accounts classmethod

list_card_transfers_for_accounts(
    session,
    /,
    account_ids,
    *,
    start_effective_date=None,
    end_effective_date=None,
)

Find all the card transfers for the given accounts and optional time period.

Note: this will include TransferEvents that are out of bounds of the given time period. If you only want the events that happened within a time period, you should use the find_by_event_date method.

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/card_transfer.py

@classmethod
def list_card_transfers_for_accounts(
    cls,
    session: Session,
    /,
    account_ids: list["AccountId"],
    *,
    start_effective_date: datetime | None = None,
    end_effective_date: datetime | None = None,
) -> list[CardTransfer]:
    """Find all the card transfers for the given accounts and optional time period.

    Note: this will include TransferEvents that are out of bounds of the given time period.
    If you only want the events that happened within a time period, you should use the
    find_by_event_date method.

    """
    return cls.search_card_transfers(
        session,
        account_ids=account_ids,
        start_effective_date=start_effective_date,
        end_effective_date=end_effective_date,
    )

list_card_transfers_for_cards classmethod

list_card_transfers_for_cards(
    session,
    /,
    card_ids,
    *,
    without_transfer_history_only=False,
)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/card_transfer.py

@classmethod
def list_card_transfers_for_cards(
    cls,
    session: Session,
    /,
    card_ids: Iterable[CardId],
    *,
    without_transfer_history_only: bool = False,
) -> list[CardTransfer]:
    query = (
        cls.select()
        .filter(CardTransfer.card_id.in_(card_ids))
        .order_by(CardTransfer.effective_date.desc())
    )
    if without_transfer_history_only:
        query = query.filter(CardTransfer.transfer_history_id.is_(None))

    return list(session.execute(query).scalars().all())

list_card_transfers_for_transfer_histories classmethod

list_card_transfers_for_transfer_histories(
    session,
    /,
    transfer_history_ids,
    *,
    start_effective_date=None,
    end_effective_date=None,
)

Find all the card transfers for the given transfer histories and optional time period.

Note: this will include TransferEvents that are out of bounds of the given time period. If you only want the events that happened within a time period, you should use the find_by_event_date method.

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/card_transfer.py

@classmethod
def list_card_transfers_for_transfer_histories(
    cls,
    session: Session,
    /,
    transfer_history_ids: list["TransferHistoryId"],
    *,
    start_effective_date: datetime | None = None,
    end_effective_date: datetime | None = None,
) -> list[CardTransfer]:
    """Find all the card transfers for the given transfer histories and optional time period.

    Note: this will include TransferEvents that are out of bounds of the given time period.
    If you only want the events that happened within a time period, you should use the
    find_by_event_date method.

    """
    return cls.search_card_transfers(
        session,
        transfer_history_ids=transfer_history_ids,
        start_effective_date=start_effective_date,
        end_effective_date=end_effective_date,
    )

list_card_transfers_for_transfer_histories_by_event_date classmethod

list_card_transfers_for_transfer_histories_by_event_date(
    session,
    /,
    transfer_history_ids,
    *,
    start_effective_date=None,
    end_effective_date=None,
    order_by_created_at=False,
)

Find all the card transfers attached to given accounts with events that happened for the given time period.

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/card_transfer.py

@classmethod
def list_card_transfers_for_transfer_histories_by_event_date(
    cls,
    session: Session,
    /,
    transfer_history_ids: list["TransferHistoryId"],
    *,
    start_effective_date: datetime | None = None,
    end_effective_date: datetime | None = None,
    order_by_created_at: bool = False,
) -> list[CardTransfer]:
    """Find all the card transfers attached to given accounts with events that happened for the given time period."""
    query = (
        cls.select()
        .distinct()
        .join(CardTransferTransferEventAssociation)
        .join(TransferEvent)
        .filter(CardTransfer.transfer_history_id.in_(transfer_history_ids))
    )

    if start_effective_date:
        query = query.filter(TransferEvent.effective_date >= start_effective_date)
    if end_effective_date:
        query = query.filter(TransferEvent.effective_date <= end_effective_date)

    if order_by_created_at:
        query = query.order_by(CardTransfer.created_at.desc())

    return list(session.execute(query).scalars().all())

model class-attribute instance-attribute

model = CardTransfer

record_card_transfer classmethod

record_card_transfer(
    session,
    /,
    *,
    effective_date,
    mcc,
    merchant_id,
    city,
    country,
    name,
    provider,
    external_id,
    card_id,
    account_id,
    postal_code=None,
    transfer_history_id=None,
    raw=None,
)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/card_transfer.py

@classmethod
def record_card_transfer(
    cls,
    session: Session,
    /,
    *,
    effective_date: datetime,
    mcc: str,
    merchant_id: str,
    city: str,
    country: str,
    name: str,
    provider: PaymentServiceProvider,
    external_id: str,
    card_id: UUID,
    account_id: UUID,
    postal_code: str | None = None,
    transfer_history_id: UUID | None = None,
    raw: dict[str, Any] | None = None,
) -> tuple[CardTransfer, bool]:
    insert_stmt = (
        insert(CardTransfer)
        .values(
            effective_date=effective_date,
            mcc=mcc,
            merchant_id=merchant_id,
            postal_code=postal_code,
            city=city,
            country=country,
            name=name,
            provider=provider,
            external_id=external_id,
            card_id=card_id,
            account_id=account_id,
            transfer_history_id=transfer_history_id,
            raw=raw,
        )
        .on_conflict_do_nothing(
            constraint="card_payment__unique_external_id_per_provider"
        )
        .returning(cls.model.id)
    )
    created = session.scalar(insert_stmt)

    # Return model from provider-specific IDs
    return session.execute(
        cls.select().filter(
            cls.model.provider == provider,
            cls.model.external_id == external_id,
        )
    ).scalar_one(), bool(created)

search_card_transfers classmethod

search_card_transfers(
    session,
    /,
    *,
    account_ids=None,
    transfer_history_ids=None,
    start_effective_date=None,
    end_effective_date=None,
    order_by_created_at=False,
)

Find all the card transfers attached to given accounts for the given time period.

Note: this will include TransferEvents that are out of bounds of the given time period. If you only want the events that happened within a time period, you should use the find_by_event_date method.

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/card_transfer.py

@classmethod
def search_card_transfers(
    cls,
    session: Session,
    /,
    *,
    account_ids: list["AccountId"] | None = None,
    transfer_history_ids: list["TransferHistoryId"] | None = None,
    start_effective_date: datetime | None = None,
    end_effective_date: datetime | None = None,
    order_by_created_at: bool = False,
) -> list[CardTransfer]:
    """Find all the card transfers attached to given accounts for the given time period.

    Note: this will include TransferEvents that are out of bounds of the given time period.
    If you only want the events that happened within a time period, you should use the
    find_by_event_date method.
    """
    query = cls.select()
    if account_ids is not None:
        query = query.filter(CardTransfer.account_id.in_(account_ids))
    if transfer_history_ids is not None:
        query = query.filter(
            CardTransfer.transfer_history_id.in_(transfer_history_ids)
        )

    if start_effective_date:
        query = query.filter(CardTransfer.effective_date >= start_effective_date)
    if end_effective_date:
        query = query.filter(CardTransfer.effective_date <= end_effective_date)

    # Build ordering
    orderings = []
    if order_by_created_at:
        orderings.append(CardTransfer.created_at.desc())
    orderings.append(CardTransfer.effective_date.desc())

    query = query.order_by(*orderings)

    return list(session.execute(query).scalars().all())

components.payment_gateway.subcomponents.transfers.models.brokers.internal_transfer

InternalTransferModelBroker

Bases: BaseModelBroker

Model broker for the InternalTransfer model.

aggregate_internal_transfer_balance classmethod

aggregate_internal_transfer_balance(
    session,
    /,
    transfer_history_id,
    *,
    effective_date,
    creation_date,
)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/internal_transfer.py

@classmethod
def aggregate_internal_transfer_balance(
    cls,
    session: Session,
    /,
    transfer_history_id: UUID,
    *,
    effective_date: datetime,
    creation_date: datetime,
) -> int:
    stmt = (
        select(
            func.sum(InternalTransfer.amount),
        )
        .where(InternalTransfer.transfer_history_id == transfer_history_id)
        .where(InternalTransfer.effective_date <= effective_date)
        .where(InternalTransfer.created_at <= creation_date)
    )
    result = session.execute(stmt).scalar()

    return result or 0

create_internal_transfer classmethod

create_internal_transfer(
    session,
    /,
    *,
    transfer_history_id,
    effective_date,
    amount,
    description,
    reference,
)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/internal_transfer.py

@classmethod
def create_internal_transfer(
    cls,
    session: Session,
    /,
    *,
    transfer_history_id: UUID,
    effective_date: datetime,
    amount: int,
    description: str,
    reference: str,
) -> InternalTransfer:
    internal_transfer = InternalTransfer(
        transfer_history_id=transfer_history_id,
        effective_date=effective_date,
        amount=amount,
        description=description,
        reference=reference,
    )
    session.add(internal_transfer)
    return internal_transfer

find_internal_transfer_by_id classmethod

find_internal_transfer_by_id(session, /, id)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/internal_transfer.py

@classmethod
def find_internal_transfer_by_id(
    cls,
    session: Session,
    /,
    id: UUID,
) -> InternalTransfer | None:
    return session.execute(
        cls.select().filter(InternalTransfer.id == id)
    ).scalar_one_or_none()

get_internal_transfer classmethod

get_internal_transfer(session, /, id)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/internal_transfer.py

@classmethod
def get_internal_transfer(
    cls,
    session: Session,
    /,
    id: UUID,
) -> InternalTransfer:
    internal_transfer: InternalTransfer = session.execute(
        cls.select().filter(InternalTransfer.id == id)
    ).scalar_one()
    return internal_transfer

list_internal_transfers_by_reference classmethod

list_internal_transfers_by_reference(session, /, reference)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/internal_transfer.py

@classmethod
def list_internal_transfers_by_reference(
    cls,
    session: Session,
    /,
    reference: str,
) -> list[InternalTransfer]:
    return list(
        session.execute(
            cls.select().filter(InternalTransfer.reference == reference)
        )
        .scalars()
        .all()
    )

list_internal_transfers_for_transfer_histories classmethod

list_internal_transfers_for_transfer_histories(
    session,
    /,
    transfer_history_ids,
    *,
    start_effective_date=None,
    end_effective_date=None,
    order_by_created_at=False,
)

Find all the internal transfers for the given transfer histories and optional time period.

Note: this will include TransferEvents that are out of bounds of the given time period. If at some point we want a function that returns only the events in the given time period, we should create a new method based on this ⧉ and not forget to add the .outerjoin.

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/internal_transfer.py

@classmethod
def list_internal_transfers_for_transfer_histories(
    cls,
    session: Session,
    /,
    transfer_history_ids: list["TransferHistoryId"],
    *,
    start_effective_date: datetime | None = None,
    end_effective_date: datetime | None = None,
    order_by_created_at: bool = False,
) -> list[InternalTransfer]:
    """Find all the internal transfers for the given transfer histories and optional time period.

    Note: this will include TransferEvents that are out of bounds of the given time period.
    If at some point we want a function that returns only the events in the given time period,
    we should create a new method based on [this](https://stackoverflow.com/questions/67578639/flask-sqlalchemy-filter-objects-in-relationship-for-each-object)
    and not forget to add the .outerjoin.

    """
    query = cls.select().filter(
        InternalTransfer.transfer_history_id.in_(transfer_history_ids)
    )
    if start_effective_date:
        query = query.filter(
            InternalTransfer.effective_date >= start_effective_date
        )
    if end_effective_date:
        query = query.filter(InternalTransfer.effective_date <= end_effective_date)

    # Build ordering
    orderings = []
    if order_by_created_at:
        orderings.append(InternalTransfer.created_at.desc())
    orderings.append(InternalTransfer.effective_date.desc())

    query = query.order_by(*orderings)

    return list(session.execute(query).scalars().all())

model class-attribute instance-attribute

model = InternalTransfer

components.payment_gateway.subcomponents.transfers.models.brokers.transfer_event

TransferEventModelBroker

Bases: BaseModelBroker

Model broker for the TransferEvent model.

get_transfer_event_by_external_id classmethod

get_transfer_event_by_external_id(
    session, /, provider, external_id
)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/transfer_event.py

@classmethod
def get_transfer_event_by_external_id(
    cls,
    session: Session,
    /,
    provider: PaymentServiceProvider,
    external_id: str,
) -> TransferEvent:
    transfer_event: TransferEvent = session.execute(
        cls.select().filter(
            cls.model.provider == provider,
            cls.model.external_id == external_id,
        )
    ).scalar_one()
    return transfer_event

model class-attribute instance-attribute

model = TransferEvent

record_transfer_event classmethod

record_transfer_event(
    session,
    /,
    effective_date,
    received,
    reserved,
    balance,
    status,
    provider,
    external_id,
    raw=None,
)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/transfer_event.py

@classmethod
def record_transfer_event(
    cls,
    session: Session,
    /,
    effective_date: datetime,
    received: int,
    reserved: int,
    balance: int,
    status: str,
    provider: PaymentServiceProvider,
    external_id: str,
    raw: dict[str, Any] | None = None,
) -> tuple[TransferEvent, bool]:
    insert_stmt = (
        insert(cls.model)
        .values(
            effective_date=effective_date,
            received=received,
            reserved=reserved,
            balance=balance,
            status=status,
            provider=provider,
            external_id=external_id,
            raw=raw,
        )
        .on_conflict_do_nothing(
            constraint="transfer_event__unique_external_id_per_provider"
        )
        .returning(cls.model.id)
    )
    created = session.scalar(insert_stmt)

    # Return model from provider-specific IDs
    return session.execute(
        cls.select().filter(
            cls.model.provider == provider,
            cls.model.external_id == external_id,
        )
    ).scalar_one(), bool(created)

components.payment_gateway.subcomponents.transfers.models.brokers.transfer_history

TransferHistoryModelBroker

Bases: BaseModelBroker

Model broker for the TransferHistory model.

create_transfer_history classmethod

create_transfer_history(
    session, /, *, private_type, private_ref
)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/transfer_history.py

@classmethod
def create_transfer_history(
    cls,
    session: Session,
    /,
    *,
    private_type: str,
    private_ref: str,
) -> TransferHistory:
    transfer_history = TransferHistory(
        private_type=private_type, private_ref=private_ref
    )
    session.add(transfer_history)
    return transfer_history

get_transfer_history classmethod

get_transfer_history(session, /, id)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/transfer_history.py

@classmethod
def get_transfer_history(
    cls,
    session: Session,
    /,
    id: UUID,
) -> TransferHistory:
    transfer_history: TransferHistory = session.execute(
        cls.select().filter(TransferHistory.id == id)
    ).scalar_one()
    return transfer_history

get_transfer_history_by_private_ref classmethod

get_transfer_history_by_private_ref(
    session, /, *, private_type, private_ref
)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/transfer_history.py

@classmethod
def get_transfer_history_by_private_ref(
    cls,
    session: Session,
    /,
    *,
    private_type: str,
    private_ref: str,
) -> TransferHistory:
    transfer_history: TransferHistory = session.execute(
        cls.select().filter(
            TransferHistory.private_type == private_type,
            TransferHistory.private_ref == private_ref,
        )
    ).scalar_one()
    return transfer_history

list_transfer_histories classmethod

list_transfer_histories(session, /, ids)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/transfer_history.py

@classmethod
def list_transfer_histories(
    cls,
    session: Session,
    /,
    ids: list[UUID],
) -> list[TransferHistory]:
    transfer_histories = (
        session.execute(cls.select().filter(TransferHistory.id.in_(ids)))
        .scalars()
        .all()
    )
    return list(transfer_histories)

model class-attribute instance-attribute

model = TransferHistory

record_transfer_history classmethod

record_transfer_history(
    session, /, *, private_type, private_ref
)

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/transfer_history.py

@classmethod
def record_transfer_history(
    cls,
    session: Session,
    /,
    *,
    private_type: str,
    private_ref: str,
) -> tuple[TransferHistory, bool]:
    insert_stmt = (
        insert(TransferHistory)
        .values(
            private_type=private_type,
            private_ref=private_ref,
        )
        .on_conflict_do_nothing(
            "transfer_history_private_key_and_ref_unique_constraint"
        )
        .returning(cls.model.id)
    )

    created = session.scalar(insert_stmt)
    return session.execute(
        cls.select().filter(
            TransferHistory.private_type == private_type,
            TransferHistory.private_ref == private_ref,
        )
    ).scalar_one(), bool(created)

components.payment_gateway.subcomponents.transfers.models.brokers.transfer_update

TransferUpdateModelBroker

Bases: BaseModelBroker

get_transfer_updates_by_bank_transfer_id classmethod

get_transfer_updates_by_bank_transfer_id(
    session, /, bank_transfer_id
)

Get all transfer updates for a given bank transfer ID, ordered by sequence number descending.

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/transfer_update.py

@classmethod
def get_transfer_updates_by_bank_transfer_id(
    cls,
    session: Session,
    /,
    bank_transfer_id: BankTransferId,
) -> list[TransferUpdate]:
    """
    Get all transfer updates for a given bank transfer ID, ordered by sequence number descending.
    """
    transfer_updates = (
        session.execute(
            cls.select()
            .filter(
                TransferUpdate.transfer_type == TransferUpdateTransferType.BANK,
                TransferUpdate.transfer_id == bank_transfer_id,
            )
            .order_by(TransferUpdate.sequence_number.desc())
        )
        .scalars()
        .all()
    )
    return list(transfer_updates)

model class-attribute instance-attribute

model = TransferUpdate

record_transfer_update classmethod

record_transfer_update(
    session,
    /,
    *,
    transfer_id,
    transfer_type,
    direction,
    sequence_number,
    occurred_at,
    amount,
    currency,
    status,
    provider,
    external_transfer_id,
    raw=None,
    external_transaction_id=None,
)

Create a transfer update if it doesn't exist, else do nothing (idempotent).

It's OK to just ignore the updated fields if the row already exists, as the PSPs should send the same information on every occurrence of the event, and there's nothing we can do with inconsistent updates. So we just record the first one and ignore the rest.

Returns:

Name	Type	Description
TransferUpdate	TransferUpdate	The transfer update model instance.
bool	bool	True if the transfer update was created, False if it already existed.

Source code in components/payment_gateway/subcomponents/transfers/models/brokers/transfer_update.py

@classmethod
def record_transfer_update(
    cls,
    session: Session,
    /,
    *,
    transfer_id: UUID,
    transfer_type: TransferUpdateTransferType,
    direction: TransferDirection,
    sequence_number: int,
    occurred_at: datetime,
    amount: int,
    currency: str,
    status: str,
    provider: PaymentServiceProvider,
    external_transfer_id: str,
    raw: dict[str, Any] | None = None,
    external_transaction_id: str | None = None,
) -> tuple[TransferUpdate, bool]:
    """Create a transfer update if it doesn't exist, else do nothing (idempotent).

    It's OK to just ignore the updated fields if the row already exists, as
    the PSPs should send the same information on every occurrence of the
    event, and there's nothing we can do with inconsistent updates. So we
    just record the first one and ignore the rest.

    Returns:
        TransferUpdate: The transfer update model instance.
        bool: True if the transfer update was created, False if it already existed.
    """
    insert_stmt = (
        insert(cls.model)
        .values(
            transfer_id=transfer_id,
            transfer_type=transfer_type,
            direction=direction,
            sequence_number=sequence_number,
            occurred_at=occurred_at,
            amount=amount,
            currency=currency,
            status=status,
            provider=provider,
            raw=raw,
            external_transfer_id=external_transfer_id,
            external_transaction_id=external_transaction_id,
        )
        .on_conflict_do_nothing(
            constraint="transfer_update__unique_provider_specific_ids"
        )
        .returning(cls.model.id)
    )
    created = session.scalar(insert_stmt)

    # Return model from provider-specific IDs
    return session.execute(
        cls.select().filter(
            cls.model.provider == provider,
            cls.model.external_transfer_id == external_transfer_id,
            cls.model.sequence_number == sequence_number,
        )
    ).scalar_one(), bool(created)