m-ledger Module

Port/Adapter abstraction for ledger operations. Default: Book-Keeper. Future: Stock-Keeper.

---

Protocol

from typing import Protocol
from decimal import Decimal
from datetime import date

class LedgerProtocol(Protocol):
    """Abstract ledger interface — swap implementations without code changes."""

    # Account Management
    async def create_account(
        self,
        code: str,
        name: str,
        account_type: str,  # Asset, Liability, Income, Expense, Equity
        parent: str | None = None,
    ) -> str:
        """Create ledger account. Returns account ID."""
        ...

    async def get_account(self, code: str) -> dict | None:
        """Get account by code."""
        ...

    # Entries
    async def post_entry(
        self,
        debit_account: str,
        credit_account: str,
        amount: Decimal,
        narration: str,
        voucher_type: str,
        voucher_no: str,
        posting_date: date,
        metadata: dict | None = None,
    ) -> str:
        """Post double-entry. Returns entry ID."""
        ...

    async def reverse_entry(self, entry_id: str, reason: str) -> str:
        """Reverse an existing entry."""
        ...

    # Queries
    async def get_balance(
        self,
        account: str,
        as_of: date | None = None,
    ) -> Decimal:
        """Get account balance as of date."""
        ...

    async def get_statement(
        self,
        account: str,
        from_date: date,
        to_date: date,
    ) -> list[dict]:
        """Get account statement with running balance."""
        ...

    async def get_trial_balance(self, as_of: date) -> list[dict]:
        """Trial balance for all accounts."""
        ...

---

Adapters

BookKeeperAdapter (Default)

class BookKeeperAdapter(LedgerProtocol):
    """Uses Book-Keeper's TigerBeetle-based ledger."""

    def __init__(self, client: BookKeeperClient):
        self.client = client

    async def post_entry(self, debit_account: str, credit_account: str, ...):
        return await self.client.create_transfer(
            debit_account_id=await self._resolve_account(debit_account),
            credit_account_id=await self._resolve_account(credit_account),
            amount=int(amount * 100),  # TigerBeetle uses integers
            ...
        )

    async def get_balance(self, account: str, as_of: date | None = None):
        account_id = await self._resolve_account(account)
        result = await self.client.get_account_balance(account_id)
        return Decimal(result.balance) / 100

SQLLedgerAdapter (Fallback)

class SQLLedgerAdapter(LedgerProtocol):
    """Pure SQL ledger for simple deployments."""

    async def post_entry(self, debit_account: str, credit_account: str, ...):
        async with self.db.transaction():
            entry_id = await self.db.execute(
                "INSERT INTO ledger_entry (...) VALUES (...) RETURNING id",
                ...
            )
            await self.db.execute(
                "INSERT INTO ledger_line (entry_id, account, debit, credit) VALUES ...",
                entry_id, debit_account, amount, 0
            )
            await self.db.execute(
                "INSERT INTO ledger_line (entry_id, account, debit, credit) VALUES ...",
                entry_id, credit_account, 0, amount
            )
        return entry_id

StockKeeperAdapter (Future)

class StockKeeperAdapter(LedgerProtocol):
    """Dedicated stock ledger for high-throughput inventory."""
    # Same interface, optimized for stock movements
    # Uses TigerBeetle with stock-specific account structure

Implementation Rules for LLMs

When building integrations for m-ledger or stock-keeper, AI Agents must adhere to the following:

1. Book-Keeper is an INTERNAL Microservice:

   • book-keeper lives in this repository under apps/book-keeper.
   • Interactions must happen via standard HTTP POST REST API calls (e.g. POST /api/book-keeper/v1/journal-entries).
   • Every request must include the tenant_id of the consuming application.
   • All complex double-entry financial logic (like enforcing limits, deducting balances, creating suspense accounts) happens via atomic JSON transfer payloads defining debit_legs and credit_legs.

2. Stock-Keeper is a LOCAL Microservice:

   • stock-keeper lives in this repository under apps/stock-keeper.
   • It is a pure Litestar application implementing a minimal Hexagonal Architecture (Ports and Adapters), built for maximum performance.
   • It is NOT built with Framework M. It directly interfaces with TigerBeetle for high-throughput inventory ledgers.

Example HTTP Book-Keeper Post:

import httpx

async def submit_journal_entry(tenant_id: str, debit_account: str, credit_account: str, amount: int):
    payload = {
        "tenant_id": tenant_id,
        "entry_date": "2024-08-01",
        "narration": "Automated transfer",
        "debit_legs": [{"account_code": debit_account, "amount": amount, "currency": "INR"}],
        "credit_legs": [{"account_code": credit_account, "amount": amount, "currency": "INR"}]
    }
    async with httpx.AsyncClient() as client:
        response = await client.post("http://book-keeper:8000/api/book-keeper/v1/journal-entries", json=payload)
        response.raise_for_status()

---

Configuration

config.py

from enum import Enum

class LedgerBackend(str, Enum):
    BOOK_KEEPER = "book_keeper"
    HTTP_LOCAL = "http_local"  # Used for local Stock-Keeper microservice
    SQL = "sql"

# settings.py
LEDGER_BACKEND: LedgerBackend = LedgerBackend.BOOK_KEEPER
BOOK_KEEPER_URL: str = "http://localhost:8080"

# dependency injection
def get_ledger() -> LedgerProtocol:
    match settings.LEDGER_BACKEND:
        case LedgerBackend.BOOK_KEEPER:
            return BookKeeperAdapter(BookKeeperClient(settings.BOOK_KEEPER_URL))
        case LedgerBackend.SQL:
            return SQLLedgerAdapter(get_db())
        case LedgerBackend.STOCK_KEEPER:
            return StockKeeperAdapter(StockKeeperClient(settings.STOCK_KEEPER_URL))

---

Usage in Apps

from m_ledger import get_ledger

async def post_purchase_receipt(receipt: PurchaseReceipt):
    ledger = get_ledger()

    # Post to GL: Debit Stock, Credit Supplier
    await ledger.post_entry(
        debit_account="1200-STOCK",
        credit_account=f"2100-SUPPLIER-{receipt.supplier}",
        amount=receipt.total_amount,
        voucher_type="PurchaseReceipt",
        voucher_no=receipt.receipt_no,
        posting_date=receipt.posting_date,
    )

---

Events Published

Event	When
ledger.entry.created	New entry posted
ledger.entry.reversed	Entry reversed
ledger.account.created	New account created