Queries

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

LedgerQueries

This class contains all the queries related to ledgers.

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
session	Session	The Session to use to make requests.	required
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/protected/business_logic/queries/ledger_queries.py

@obs.api_call()
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:
        session: The Session to use to make requests.
        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_entry

get_entry(session, /, id)

Get a ledger entry entity from its ID.

Parameters:

Name	Type	Description	Default
session	Session	The Session to use to make requests.	required
id	LedgerEntryId	The ID of the ledger entry.	required

Source code in components/payment_gateway/subcomponents/ledgers/protected/business_logic/queries/ledger_queries.py

@obs.api_call()
def get_entry(
    self,
    session: Session,
    /,
    id: LedgerEntryId,
) -> LedgerEntry:
    """Get a ledger entry entity from its ID.

    Args:
        session: The Session to use to make requests.
        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

get_ledger(session, /, id)

Get a ledger entity from its ID.

Parameters:

Name	Type	Description	Default
session	Session	The Session to use to make requests.	required
id	LedgerId	The ID of the ledger.	required

Source code in components/payment_gateway/subcomponents/ledgers/protected/business_logic/queries/ledger_queries.py

@obs.api_call()
def get_ledger(
    self,
    session: Session,
    /,
    id: LedgerId,
) -> Ledger:
    """Get a ledger entity from its ID.

    Args:
        session: The Session to use to make requests.
        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_balance

get_ledger_balance(session, /, id, effective_at=None)

Get current balance of a ledger.

This is the ending balance of the entry that is effective at the given date, or the last entry if no date is given.

Parameters:

Name	Type	Description	Default
session	Session	The session to use for the database operations.	required
id	LedgerId	The ID of the ledger.	required
effective_at	datetime | None	The point in time to get the ledger balance for. None will use the last entry.	None

Source code in components/payment_gateway/subcomponents/ledgers/protected/business_logic/queries/ledger_queries.py

@obs.api_call()
def get_ledger_balance(
    self,
    session: Session,
    /,
    id: LedgerId,
    effective_at: datetime | None = None,
) -> int:
    """Get current balance of a ledger.

    This is the ending balance of the entry that is effective at the given date, or the last entry if no date is given.

    Args:
        session: The session to use for the database operations.
        id: The ID of the ledger.
        effective_at: The point in time to get the ledger balance for. `None` will use the last entry.
    """
    ledger_entry = self._get_ledger_entry_effective_at(
        session,
        id,
        effective_at,
    )
    return ledger_entry.ending_balance if ledger_entry else 0

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
session	Session	The Session to use to make requests.	required
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/protected/business_logic/queries/ledger_queries.py

@obs.api_call()
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:
        session: The Session to use to make requests.
        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.
    """
    ledger_entry = self._get_ledger_entry_effective_at(
        session, ledger_id=ledger_id, effective_at=effective_at
    )

    return _to_ledger_entry(ledger_entry) if ledger_entry else None

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
session	Session	The Session to use to make requests.	required
reference	str	The reference of the ledger(s).	required

Source code in components/payment_gateway/subcomponents/ledgers/protected/business_logic/queries/ledger_queries.py

@obs.api_call()
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:
        session: The Session to use to make requests.
        reference: The reference of the ledger(s).
    """
    return [
        LedgerId(ledger_id)
        for ledger_id in LedgerModelBroker.list_ledger_ids_by_reference(
            session, reference=reference
        )
    ]