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¶

LedgerActions ¶

This class contains all the actions related to ledgers.

Functions¶

create_ledger ¶

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

Create a ledger.

Parameters:

Name	Type	Description	Default
`session`	`Session`	The Session to use to make requests.	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/actions/ledger_actions.py

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

    Args:
        session: The Session to use to make requests.
        description: The description of the ledger.
        reference: An optional reference to the ledger.
    """
    ledger = LedgerModelBroker.create_ledger(
        session,
        description=description,
        reference=reference,
    )
    return LedgerId(ledger.id)

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.

This entry will become the latest entry in the ledger. This operation performs the required bookkeeping on balances in the process.

Parameters:

Name	Type	Description	Default
`session`	`Session`	The Session to use to make requests.	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/actions/ledger_actions.py

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.

    This entry will become the latest entry in the ledger. This operation
    performs the required bookkeeping on balances in the process.

    Args:
        session: The Session to use to make requests.
        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.
    """
    with raise_if_ledger_not_found(id):
        ledger = LedgerModelBroker.get_ledger(session, id=id)

    raise_on_terminated_ledger(ledger)

    ledger_entry = LedgerEntryModelBroker.create_ledger_entry(
        session,
        ledger_id=ledger.id,
        opening_balance=ledger.balance,
        amount=amount,
        ending_balance=ledger.balance + amount,
        occurred_at=occurred_at,
        description=description,
        reference=reference,
        entry_metadata=metadata,
        external_transaction_id=external_transaction_id,
    )
    return LedgerEntryId(ledger_entry.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 param. ⚠️ WARNING: This method should be used carefully, as it can break the linear history of a ledger if not used carefully

Should only be called from the LedgerActions.record_entry_overwriting_created_at method.

Parameters:

Name	Type	Description	Default
`session`	`Session`	The Session to use to make requests.	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`
`created_at`	`datetime`	An optional time at which the entry was created. Allows to write an entry in the past and recomputes all entries past this date. This param should only be passed from the record_entry_overwriting_created_at method.	required

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

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 param.
    ⚠️ WARNING: This method should be used carefully, as it can break the linear
        history of a ledger if not used carefully

    Should only be called from the LedgerActions.record_entry_overwriting_created_at method.

    Args:
        session: The Session to use to make requests.
        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: An optional time at which the entry was created. Allows to write an entry in the past and recomputes
            all entries past this date. This param should only be passed from the record_entry_overwriting_created_at method.
    """
    with raise_if_ledger_not_found(id):
        ledger = LedgerModelBroker.get_ledger(session, id=id)

    raise_on_terminated_ledger(ledger)

    previous_ledger_entry = LedgerEntryModelBroker.find_ledger_entry_effective_at(
        session, ledger_id=id, effective_at=created_at
    )

    previous_balance = (
        previous_ledger_entry.ending_balance if previous_ledger_entry else 0
    )
    balance_change = amount

    ledger_entries_to_modify_balance = [
        entry for entry in ledger.entries if entry.created_at > created_at
    ]

    ledger_entries_to_overwrite = [
        entry for entry in ledger.entries if entry.created_at == created_at
    ]

    if len(ledger_entries_to_overwrite) == 0:
        ledger_entry = LedgerEntryModelBroker.create_ledger_entry_at(
            session,
            ledger_id=ledger.id,
            opening_balance=previous_balance,
            amount=balance_change,
            ending_balance=previous_balance + balance_change,
            occurred_at=occurred_at,
            description=description,
            reference=reference,
            entry_metadata=metadata,
            external_transaction_id=external_transaction_id,
            created_at=created_at,
        )

    # In case there are entries already at the created_at date, what we do is
    # modifying the balance instead of creating a new entry with the same date, which can be problematic
    for ledger_entry in ledger_entries_to_overwrite:
        balance_change -= ledger_entry.amount
        LedgerEntryModelBroker.modify_entry_balance(
            session,
            ledger_id=ledger_entry.id,
            balance_change=balance_change,
        )

    for entry in ledger_entries_to_modify_balance:
        LedgerEntryModelBroker.modify_entry_balance(
            session, ledger_id=entry.id, balance_change=balance_change
        )

    return LedgerEntryId(ledger_entry.id)

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 to make requests.	required
`id`	`LedgerId`	The ID of the ledger to terminate.	required

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

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 to make requests.
        id: The ID of the ledger to terminate.
    """
    with raise_if_ledger_not_found(id):
        ledger = LedgerModelBroker.get_ledger(session, id=id)

    if not ledger.is_terminated:
        LedgerModelBroker.terminate_ledger(session, id=id)

LedgerEntryNotFoundException ¶

Bases: PaymentLedgerException

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

LedgerNotFoundException ¶

Bases: PaymentLedgerException

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

LedgerQueries ¶

This class contains all the queries related to ledgers.

Functions¶

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
        )
    ]

LedgerTerminatedException ¶

Bases: PaymentLedgerException

Exception raised when trying to use a terminated Ledger.

PaymentLedgerException ¶

Bases: PaymentGatewayException

Base class for all Ledger exceptions.

components.payment_gateway.public.ledgers ¶

Attributes¶

LedgerId module-attribute ¶

Classes¶

LedgerActions ¶

Functions¶

create_ledger ¶

record_entry ¶

record_entry_overwriting_created_at ¶

terminate_ledger ¶

LedgerEntryNotFoundException ¶

LedgerNotFoundException ¶

LedgerQueries ¶

Functions¶

find_ledger_entries_by_reference_prefix ¶

get_entry ¶

get_ledger ¶

get_ledger_balance ¶

get_ledger_entry_effective_at ¶

get_ledger_ids_by_reference ¶

LedgerTerminatedException ¶

PaymentLedgerException ¶

LedgerId `module-attribute` ¶