Skip to content

Queries

components.payment_gateway.subcomponents.ledgers.business_logic.queries.ledger_entry_queries

LedgerEntryQueries

This class contains all the queries related to ledger entries.

find_ledger_entries_by_reference_prefix

find_ledger_entries_by_reference_prefix(
    session, /, ledger_id, reference_prefix
)

Get all the ledger entries for a reference prefix.

Parameters:

Name Type Description Default
ledger_id LedgerId

The ID of the ledger.

required
reference_prefix str

The reference prefix of the ledger entries.

required
Source code in components/payment_gateway/subcomponents/ledgers/business_logic/queries/ledger_entry_queries.py
def find_ledger_entries_by_reference_prefix(
    self,
    session: Session,
    /,
    ledger_id: LedgerId,
    reference_prefix: str,
) -> list[LedgerEntry]:
    """Get all the ledger entries for a reference prefix.

    Args:
        ledger_id: The ID of the ledger.
        reference_prefix: The reference prefix of the ledger entries.
    """
    return [
        _to_ledger_entry(entry)
        for entry in LedgerEntryModelBroker.find_ledger_entries_by_reference_prefix(
            session, ledger_id, reference_prefix
        )
    ]

get_ledger_entry

get_ledger_entry(session, /, id)

Get a ledger entry entity from its ID.

Parameters:

Name Type Description Default
id LedgerEntryId

The ID of the ledger entry.

required
Source code in components/payment_gateway/subcomponents/ledgers/business_logic/queries/ledger_entry_queries.py
def get_ledger_entry(
    self,
    session: Session,
    /,
    id: LedgerEntryId,
) -> LedgerEntry:
    """Get a ledger entry entity from its ID.

    Args:
        id: The ID of the ledger entry.
    """
    with raise_if_ledger_entry_not_found(id):
        ledger_entry = LedgerEntryModelBroker.get_ledger_entry(session, id=id)
        return _to_ledger_entry(ledger_entry)

get_ledger_entry_effective_at

get_ledger_entry_effective_at(
    session, /, ledger_id, effective_at
)

Get the ledger entry for a given ledger at a certain point in time.

The logic is based on entries' created_at and not the occured_at time to ensure accurate time travel. We want to know same ledger balance that we would have seen at that time.

Parameters:

Name Type Description Default
ledger_id LedgerId

The ID of the ledger.

required
effective_at datetime | None

The point in time to get the ledger entry for. None will return the last entry.

required
Source code in components/payment_gateway/subcomponents/ledgers/business_logic/queries/ledger_entry_queries.py
def get_ledger_entry_effective_at(
    self,
    session: Session,
    /,
    ledger_id: LedgerId,
    effective_at: datetime | None,
) -> LedgerEntry | None:
    """Get the ledger entry for a given ledger at a certain point in time.

    The logic is based on entries' `created_at` and not the `occured_at`
    time to ensure accurate time travel. We want to know same ledger balance
    that we would have seen at that time.

    Args:
        ledger_id: The ID of the ledger.
        effective_at: The point in time to get the ledger entry for. `None` will return the last entry.
    """

    with raise_if_ledger_not_found(ledger_id):
        ledger = LedgerModelBroker.get_ledger(session, id=ledger_id)

    raise_on_terminated_ledger(ledger)

    ledger_entry: LedgerEntryModel | None
    if effective_at is None:
        ledger_entry = ledger.last_entry
    else:
        ledger_entry = LedgerEntryModelBroker.find_ledger_entry_effective_at(
            session, ledger_id=ledger.id, effective_at=effective_at
        )
    return _to_ledger_entry(ledger_entry) if ledger_entry else None

components.payment_gateway.subcomponents.ledgers.business_logic.queries.ledger_queries

LedgerQueries

This class contains all the queries related to ledgers.

get_ledger

get_ledger(session, /, id)

Get a ledger entity from its ID.

Parameters:

Name Type Description Default
id LedgerId

The ID of the ledger.

required
Source code in components/payment_gateway/subcomponents/ledgers/business_logic/queries/ledger_queries.py
def get_ledger(
    self,
    session: Session,
    /,
    id: LedgerId,
) -> Ledger:
    """Get a ledger entity from its ID.

    Args:
        id: The ID of the ledger.
    """
    with raise_if_ledger_not_found(id):
        ledger = LedgerModelBroker.get_ledger(session, id=id)

    raise_on_terminated_ledger(ledger)

    return Ledger(
        balance=ledger.balance,
        description=ledger.description,
        reference=ledger.reference,
    )

get_ledger_ids_by_reference

get_ledger_ids_by_reference(session, /, reference)

Get all the ledger IDs for a reference.

Returns 0-n ledger IDs because nothing prevents a reference from being shared between multiple ledgers (this is a business concern and there's no unicity constraint).

Parameters:

Name Type Description Default
reference str

The reference of the ledger(s).

required
Source code in components/payment_gateway/subcomponents/ledgers/business_logic/queries/ledger_queries.py
def get_ledger_ids_by_reference(
    self,
    session: Session,
    /,
    reference: str,
) -> list[LedgerId]:
    """Get all the ledger IDs for a reference.

    Returns 0-n ledger IDs because nothing prevents a reference from being
    shared between multiple ledgers (this is a business concern and there's
    no unicity constraint).

    Args:
        reference: The reference of the ledger(s).
    """
    return [
        LedgerId(ledger_id)
        for ledger_id in LedgerModelBroker.list_ledger_ids_by_reference(
            session, reference=reference
        )
    ]