components.payment_gateway.public.ledgers

This module defines the public API for the ledgers subcomponent.

Only business logic is exposed here. Basic entities and enums are exposed in separate modules to avoid loading the entire subcomponent with its models and dependencies when they are not needed.

Attributes

LedgerId module-attribute

LedgerId = NewType('LedgerId', UUID)

Classes

LedgerEntryNotFoundException

Bases: PaymentLedgerException

Exception raised when trying to use a non-existing Ledger Entry.

LedgerLogic

LedgerLogic()

This class is the public interface to the ledger logic.

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

def __init__(self) -> None:
    self.ledger_queries = LedgerQueries()
    self.ledger_actions = LedgerActions()
    self.ledger_entry_queries = LedgerEntryQueries()

Attributes

ledger_actions instance-attribute

ledger_actions = LedgerActions()

ledger_entry_queries instance-attribute

ledger_entry_queries = LedgerEntryQueries()

ledger_queries instance-attribute

ledger_queries = LedgerQueries()

Functions

create_ledger

create_ledger(session, /, description, reference=None)

Create a ledger.

Parameters:

Name	Type	Description	Default
session	Session	The session to use for the database operations.	required
description	str	The description of the ledger.	required
reference	str | None	An optional reference to the ledger.	None

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

@obs.api_call()
def create_ledger(
    self,
    session: Session,
    /,
    description: str,
    reference: str | None = None,
) -> LedgerId:
    """Create a ledger.

    Args:
        session: The session to use for the database operations.
        description: The description of the ledger.
        reference: An optional reference to the ledger.
    """
    return self.ledger_actions.create_ledger(
        session,
        description=description,
        reference=reference,
    )

get_entries_by_reference_prefix

get_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 for the database operations.	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/ledgers.py

@obs.api_call()
def get_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 for the database operations.
        ledger_id: The ID of the ledger.
        reference_prefix: The reference prefix of the ledger entries.
    """
    return self.ledger_entry_queries.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 for the database operations.	required
id	LedgerEntryId	The ID of the ledger entry.	required

Source code in components/payment_gateway/subcomponents/ledgers/protected/business_logic/ledgers.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 for the database operations.
        id: The ID of the ledger entry.
    """
    return self.ledger_entry_queries.get_ledger_entry(
        session,
        id,
    )

get_ledger

get_ledger(session, /, id)

Get a ledger entity from its ID.

Parameters:

Name	Type	Description	Default
session	Session	The session to use for the database operations.	required
id	LedgerId	The ID of the ledger.	required

Source code in components/payment_gateway/subcomponents/ledgers/protected/business_logic/ledgers.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 for the database operations.
        id: The ID of the ledger.
    """
    return self.ledger_queries.get_ledger(
        session,
        id,
    )

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/ledgers.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.ledger_entry_queries.get_ledger_entry_effective_at(
        session,
        id,
        effective_at,
    )
    return ledger_entry.ending_balance if ledger_entry else 0

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 for the database operations.	required
reference	str	The reference of the ledger(s).	required

Source code in components/payment_gateway/subcomponents/ledgers/protected/business_logic/ledgers.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 for the database operations.
        reference: The reference of the ledger(s).
    """
    return self.ledger_queries.get_ledger_ids_by_reference(
        session,
        reference,
    )

record_entry

record_entry(
    session,
    /,
    id,
    amount,
    occurred_at,
    description=None,
    reference=None,
    metadata=None,
    external_transaction_id=None,
)

Record a new entry in a ledger.

Parameters:

Name	Type	Description	Default
session	Session	The session to use for the database operations.	required
id	LedgerId	The ID of the ledger.	required
amount	int	The amount of the entry.	required
occurred_at	datetime	The time the event occurred (not the time we process it).	required
description	str | None	An optional description of the entry.	None
reference	str | None	An optional reference to the entry.	None
metadata	dict | None	An optional metadata dictionary.	None
external_transaction_id	str | None	An optional external transaction ID. Useful for recording transactions from external systems.	None

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

@obs.api_call()
def record_entry(
    self,
    session: Session,
    /,
    id: LedgerId,
    amount: int,
    occurred_at: datetime,
    description: str | None = None,
    reference: str | None = None,
    metadata: dict | None = None,  # type: ignore[type-arg]
    external_transaction_id: str | None = None,
) -> LedgerEntryId:
    """Record a new entry in a ledger.

    Args:
        session: The session to use for the database operations.
        id: The ID of the ledger.
        amount: The amount of the entry.
        occurred_at: The time the event occurred (not the time we process it).
        description: An optional description of the entry.
        reference: An optional reference to the entry.
        metadata: An optional metadata dictionary.
        external_transaction_id: An optional external transaction ID. Useful for recording transactions from external systems.
    """
    return self.ledger_actions.record_entry(
        session,
        id=id,
        amount=amount,
        occurred_at=occurred_at,
        description=description,
        reference=reference,
        metadata=metadata,
        external_transaction_id=external_transaction_id,
    )

record_entry_overwriting_created_at

record_entry_overwriting_created_at(
    session,
    /,
    id,
    amount,
    occurred_at,
    created_at,
    description=None,
    reference=None,
    metadata=None,
    external_transaction_id=None,
)

Record a new entry in a ledger overwriting the created_at. ⚠️ WARNING: This method should be used carefully, as it can break the linear history of a ledger if not used carefully Args: session: The session to use for the database operations. id: The ID of the ledger. amount: The amount of the entry. occurred_at: The time the event occurred (not the time we process it). description: An optional description of the entry. reference: An optional reference to the entry. metadata: An optional metadata dictionary. external_transaction_id: An optional external transaction ID. Useful for recording transactions from external systems. created_at: The time the entry was created.

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

@obs.api_call()
def record_entry_overwriting_created_at(
    self,
    session: Session,
    /,
    id: LedgerId,
    amount: int,
    occurred_at: datetime,
    created_at: datetime,
    description: str | None = None,
    reference: str | None = None,
    metadata: dict | None = None,  # type: ignore[type-arg]
    external_transaction_id: str | None = None,
) -> LedgerEntryId:
    """Record a new entry in a ledger overwriting the created_at.
        ⚠️ WARNING: This method should be used carefully, as it can break the linear
        history of a ledger if not used carefully
    Args:
        session: The session to use for the database operations.
        id: The ID of the ledger.
        amount: The amount of the entry.
        occurred_at: The time the event occurred (not the time we process it).
        description: An optional description of the entry.
        reference: An optional reference to the entry.
        metadata: An optional metadata dictionary.
        external_transaction_id: An optional external transaction ID. Useful for recording transactions from external systems.
        created_at: The time the entry was created.
    """
    return self.ledger_actions.record_entry_overwriting_created_at(
        session,
        id=id,
        amount=amount,
        occurred_at=occurred_at,
        description=description,
        reference=reference,
        metadata=metadata,
        external_transaction_id=external_transaction_id,
        created_at=created_at,
    )

terminate_ledger

terminate_ledger(session, /, id)

Terminate a ledger.

The operation is idempotent, i.e. it has no effect on already terminated entities.

Ledgers in terminal state cannot be modified or used anymore. Any attempt to use or retrieve a terminated ledger will raise a LedgerTerminatedException.

Parameters:

Name	Type	Description	Default
session	Session	The session to use for the database operations.	required
id	LedgerId	The ID of the ledger to terminate.	required

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

@obs.api_call()
def terminate_ledger(
    self,
    session: Session,
    /,
    id: LedgerId,
) -> None:
    """Terminate a ledger.

    The operation is idempotent, i.e. it has no effect on already terminated
    entities.

    Ledgers in terminal state cannot be modified or used anymore. Any
    attempt to use or retrieve a terminated ledger will raise a
    `LedgerTerminatedException`.

    Args:
        session: The session to use for the database operations.
        id: The ID of the ledger to terminate.
    """
    self.ledger_actions.terminate_ledger(
        session,
        id,
    )

LedgerNotFoundException

Bases: PaymentLedgerException

Exception raised when trying to use a non-existing Ledger.

LedgerTerminatedException

Bases: PaymentLedgerException

Exception raised when trying to use a terminated Ledger.

PaymentLedgerException

Bases: PaymentGatewayException

Base class for all Ledger exceptions.