lib-ledger-core

lib-ledger-core is an extensible, asynchronous Python library that provides standard domain models, ports, and storage adapters for immutable double-entry accounting and event-sourced state tracking.

Built around the Ports and Adapters (Hexagonal Architecture) design pattern, lib-ledger-core decouples high-level business rules from storage drivers—allowing you to easily swap underlying databases, log engines, or external financial backends without altering domain logic.

System Philosophy: CQRS & Event Sourcing

At its core, lib-ledger-core enforces a clean Command-Query Responsibility Segregation (CQRS) pattern coupled with Event Sourcing:

flowchart TD
    A["Incoming Instruction"] --> B

    subgraph Command ["1. COMMAND / VALIDATION SIDE (Ledger Backend)"]
        direction TB
        B["Accepts TransferCommand Instructions"] --> C["Validates Domain Invariants<br><i>(Account Verification, Double-Entry Rules)</i>"]
        C --> D["Commits State Transition & Returns Result"]
    end

    D -->|"Valid Transfer Result"| E

    subgraph Query ["2. QUERY / READ SIDE (Event Store Backend)"]
        direction TB
        E["Receives Validated Ledger Execution State"] --> F["Appends Immutable Events to Streams<br><i>(Enforces Optimistic Concurrency Control)</i>"]
        F --> G["Acts as Single Source of Truth<br><i>(Powers Projections & Aggregates)</i>"]
    end

    style Command fill:#1e1e2e,stroke:#89b4fa,stroke-width:2px,color:#cdd6f4
    style Query fill:#1e1e2e,stroke:#a6e3a1,stroke-width:2px,color:#cdd6f4
    style A fill:#313244,stroke:#f5e0dc,stroke-width:1px,color:#cdd6f4

Why Event Sourced Aggregates?

1. Schema-less Aggregate Evolution: Instead of modifying SQL relational table schemas whenever business rules change, domain aggregates (such as account balances, tenant summaries, or risk profiles) are projected directly by reading and replaying historical event streams. Adding new aggregate views requires no database migrations.
2. Complete Auditability: Traditional database updates overwrite previous state. The event store retains every historical event sequentially, providing a permanent, tamper-evident audit trail required for compliance and financial reconciliation.
3. Optimistic Concurrency Control (OCC): Event appending guarantees that concurrent attempts to modify the same stream are safely rejected if the sequence version changes unexpectedly.

Universal Balance Tracking: Use Cases

While designed to handle double-entry financial bookkeeping and warehouse inventory movements for ERP environments, the transfer command abstraction fits any system that moves units between two balance states:

• Financial & Double-Entry Bookkeeping: Manages credits and debits across general ledger accounts, revenue tracking, and accounts payable.
• Stock Keeping & Warehouse Movements: Validates and records inventory transfers between physical warehouses, storage bins, or supply chain nodes.
• Crypto & Coin Wallets: Manages balances, gas fees, and token movements between user wallets, cold storage, and hot pools.
• Carbon Credits & Offsets: Tracks issuance, transfer, and retirement of verified metric tons of carbon emissions ($tCO_2e$) between reserves and corporate accounts.
• Loyalty Points & Rewards: Handles issuance, transfers, holds, and redemptions of promotional rewards points.
• Compute & API Quotas: Controls consumption, allocation, and rate-limiting credits for multi-tenant microservices.

Core Architectural Components

1. Abstract Ports (Interfaces)

• LedgerPort (Validation & Command Side): Defines operations for evaluating transfer instructions, checking current balance snapshots, processing pending holds, and executing multi-leg compound movements.
• EventStorePort (Audit & Read Side): Defines append-only operations for persisting versioned event streams and querying stream history.

2. Data Models

• TransferCommand: Immutable instruction specifying debit/credit target accounts, transaction reference, metadata, pending status, and multi-leg transfer definitions.
• Entry: Immutable transaction record representing an individual debit or credit line item.

3. Backend Adapters

• SqlAlchemyLedger & SqlAlchemyEventStore: Relational backends utilizing async SQL engines to process transfers and maintain versioned JSON event streams.
• TigerBeetleLedger: High-throughput ledger integration utilizing TigerBeetle for low-latency balance tracking and account flags.
• KurrentEventStore: Event-sourcing integration built on top of KurrentDB (EventStoreDB) utilizing msgspec for fast serialization.

Dynamic Adapter Registry

Custom adapters can be created by implementing either the LedgerPort or EventStorePort interface. Third-party modules can register their custom drivers using Python entry-point groups (ledger_core.ledger and ledger_core.event_store). Once registered, the core library can dynamically discover and load adapters at runtime.

Exception & Error Hierarchy

• LedgerError: Base exception class for all errors generated by the library.
• InsufficientBalanceError: Raised when a transfer command violates non-negative balance constraints.
• OccError: Raised when a stream version mismatch occurs during an append operation to the event store.

Using lib-ledger-core

The lib-ledger-core library provides generic ledger primitives, event sourcing ports, dynamic adapter registration, and programmatic database migrations.

1. Database Migrations Programmatically

lib-ledger-core encapsulates its Alembic migration scripts internally. Higher-level applications like book-keeper use the run_migrations helper to initialize or upgrade the schema without maintaining duplicate SQL migration files:

import asyncio
from ledger_core.migrations import run_downgrade, run_migrations
from sqlalchemy.ext.asyncio import create_async_engine

DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/book_keeper"


async def setup_database():
    engine = create_async_engine(DATABASE_URL, future=True)

    # Run all pending ledger-core Alembic migrations
    await run_migrations(engine)

    # ... application setup ...

    await engine.dispose()


if __name__ == "__main__":
    asyncio.run(setup_database())

2. Basic Ledger & Event Store Usage

lib-ledger-core exposes LedgerPort and EventStorePort implementations (such as SqlAlchemyLedger, TigerBeetleLedger, SqlAlchemyEventStore, and KurrentEventStore).

import asyncio
from decimal import Decimal
from ledger_core.adapters.sqlalchemy import SqlAlchemyEventStore, SqlAlchemyLedger
from ledger_core.models import TransferCommand
from sqlalchemy.ext.asyncio import create_async_engine


async def main():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:", future=True)

    # Instantiate Adapters
    ledger = SqlAlchemyLedger(engine)
    event_store = SqlAlchemyEventStore(engine)

    # Seed an account with initial funds
    await ledger.seed_account(
        tenant_id="tenant_1", account="CASH", amount=Decimal("1000.00")
    )

    # Execute a Transfer
    cmd = TransferCommand(
        tenant_id="tenant_1",
        debit_account="EQUIPMENT",
        credit_account="CASH",
        amount=Decimal("250.00"),
        reference="INV-2026-001",
        description="Purchased office equipment",
    )
    transfer_id = await ledger.transfer(cmd)
    print(f"Executed Transfer ID: {transfer_id}")

    # Record Domain Event
    await event_store.append(
        tenant_id="tenant_1",
        stream_id="equipment-purchases",
        events=[
            {
                "type": "EquipmentPurchased",
                "transfer_id": transfer_id,
                "amount": "250.00",
            }
        ],
        expected_version=0,
    )

    # Check Balances
    cash_bal = await ledger.get_balance("tenant_1", "CASH")
    equipment_bal = await ledger.get_balance("tenant_1", "EQUIPMENT")
    print(f"CASH Balance: {cash_bal}")  # Outputs: 750.00
    print(f"EQUIPMENT Balance: {equipment_bal}")  # Outputs: 250.00

    await ledger.close()
    await event_store.close()


if __name__ == "__main__":
    asyncio.run(main())

3. Loading Adapters via Registry

Adapters can also be loaded dynamically using the entry-point registry:

from ledger_core import load_event_store_adapter, load_ledger_adapter
from sqlalchemy.ext.asyncio import create_async_engine

# Dynamically resolve factories using entry-point identifiers
ledger_factory = load_ledger_adapter("sqlalchemy")
event_store_factory = load_event_store_adapter("kurrent")

engine = create_async_engine("postgresql+asyncpg://...")
ledger = ledger_factory(engine)
event_store = event_store_factory("esdb://localhost:2113?tls=false")

4. Application Integration (book-keeper example)

Inside book-keeper, lib-ledger-core adapters are conditionally selected during application startup based on settings:

from ledger_core.adapters.kurrent import KurrentEventStore
from ledger_core.adapters.sqlalchemy import SqlAlchemyEventStore, SqlAlchemyLedger
from ledger_core.adapters.tigerbeetle import TigerBeetleLedger
from ledger_core.interfaces import EventStorePort, LedgerPort


def create_ledger(engine, settings) -> LedgerPort:
    if settings.ledger_type == "postgres":
        return SqlAlchemyLedger(engine)
    return TigerBeetleLedger(
        addresses=settings.tigerbeetle_addresses,
        account_namespace=settings.account_namespace,
    )


def create_event_store(engine, settings) -> EventStorePort:
    if settings.event_store_type == "postgres":
        return SqlAlchemyEventStore(engine)
    return KurrentEventStore(connection_string=settings.kurrent_connection_string)

License

Apache 2.0