Actions

components.payment_gateway.subcomponents.ledgers.business_logic.actions.ledger_actions ¶

LedgerActions ¶

This class contains all the actions related to ledgers.

create_ledger ¶

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

Create a ledger.

Parameters:

Name	Type	Description	Default
`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/business_logic/actions/ledger_actions.py

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

    Args:
        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
`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/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:
        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 LedgerLogic.record_entry_overwriting_created_at method.

Parameters:

Name	Type	Description	Default
`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/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 LedgerLogic.record_entry_overwriting_created_at method.

    Args:
        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.
    """
    from components.payment_gateway.public.ledgers import (
        LedgerLogic,
    )

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

    raise_on_terminated_ledger(ledger)

    ledger_logic = LedgerLogic()
    previous_balance = ledger_logic.get_ledger_balance(
        session,
        id=ledger.id,  # type: ignore[arg-type]
        effective_at=created_at,
    )
    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
`id`	`LedgerId`	The ID of the ledger to terminate.	required

Source code in components/payment_gateway/subcomponents/ledgers/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:
        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)