Ledger events¶

Events are named, user-defined operations on a ledger. Each event compiles into a single PostgreSQL function (LANGUAGE sql) that inserts rows into the ledger view and returns the inserted rows via RETURNING *.

Import from the ledger extension:

from codegen_database.ext.ledger import (
    LedgerEvent,
    construct_ledger_balances_query,
)

Reconciliation (diff mode)¶

The primary pattern is reconciliation: you have a transactional object (invoices, shipments, work orders) with line items, and you want the ledger to reflect the current state of those line items. The system handles the arithmetic — it diffs the desired state against existing balances and inserts only the correcting deltas.

This makes the operation idempotent: calling the same event twice with the same input produces zero new rows the second time. Amending the source and re-calling inserts exactly the correcting entries.

Example — revenue recognition from invoice line items:

from sqlalchemy import Integer, String, func, select
from sqlalchemy.dialects.postgresql import ARRAY

from codegen_database.ext.ledger import (
    LedgerEvent,
    construct_ledger_balances_query,
)

# Assume `invoice_lines` is a SQLAlchemy Table with columns:
#   id, invoice_id, account, amount

recognize = LedgerEvent(
    name="recognize",
    input=lambda p: select(
        func.unnest(p("invoice_ids", ARRAY(Integer)))
        .label("invoice_id"),
    ),
    desired=lambda pginput: select(
        invoice_lines.c.invoice_id,
        invoice_lines.c.account,
        invoice_lines.c.amount.label("value"),
    ).where(
        invoice_lines.c.invoice_id.in_(
            select(pginput.c.invoice_id)
        )
    ),
    existing=construct_ledger_balances_query("invoice_id", "account"),
    diff_keys=["invoice_id", "account"],
)

The generated SQL uses CTEs: input (the function parameters), desired (the target state from the source table), existing (negated current balances), and deltas (the union of desired + existing, filtered to non-zero values). This is a single INSERT ... RETURNING * statement.

Key parameters:

input: Lambda (p) -> Select that builds the input CTE. p is a ParamCollector — each call to p("name", Type) records a function parameter and returns a literal_column("p_name") reference.
desired: Lambda (pginput) -> Select that builds the desired-state CTE. pginput is a synthetic table reference to the input CTE. Typically joins against a source table (e.g. invoice_lines) to produce the ledger entries.
existing: Lambda (table, desired, pginput) -> Select that returns the negated current balances. pginput is the same input-CTE reference passed to desired, so function-param values are reached as pginput.c.<name> – no p_ prefix. Use construct_ledger_balances_query() or construct_ledger_scoped_balances_query() for the common patterns.
diff_keys: Column names used for grouping. Required when desired is set.

The construct_ledger_balances_query helper produces an existing callable that groups SUM(value) * -1 by the specified keys, filtered to only groups present in the desired CTE.

Owner-scoped reconciliation¶

construct_ledger_balances_query filters existing to rows whose diff-key tuple appears in desired. That works when every key the event might want to unwind is guaranteed to also be in desired – e.g. invoice recognition where each line item always maps to exactly one (invoice_id, account) pair.

It does not work when a state transition can retire a key. If an event posts in_transit while a lot is in flight, then re-posts under active once delivered, the in_transit row no longer appears in desired after delivery – so the default helper can’t see it, and the unwinding entry never gets emitted.

For that pattern, use construct_ledger_scoped_balances_query(), which filters existing by an arbitrary predicate over the ledger root instead:

from codegen_database.ext.ledger import construct_ledger_scoped_balances_query

def owned_by_input_po(ledger, pginput):
    return ledger.c.lot_id.in_(
        select(PurchaseLot.id)
        .join(LineItem, LineItem.id == PurchaseLot.line_item_id)
        .where(
            LineItem.purchase_order_id
            == pginput.c.purchase_order_id,
        )
    )

post_po = LedgerEvent(
    name="post_po",
    input=...,
    desired=...,
    existing=construct_ledger_scoped_balances_query(
        "lot_id", "class_id", "account", "direction",
        where=owned_by_input_po,
    ),
    diff_keys=["lot_id", "class_id", "account", "direction"],
)

The where predicate receives the ledger root table and the pginput CTE reference (same as desired), so function parameters are reached as pginput.c.<name> – no p_ prefix. Reach the owner directly via a column when it’s denormalized on the ledger, or via a subquery / join when it’s not.

Usage¶

-- Recognize revenue for two invoices:
SELECT * FROM ops.ops_revenue_recognize(
    p_invoice_ids => ARRAY[1001, 1002]
);

-- Call again — idempotent, returns 0 rows:
SELECT * FROM ops.ops_revenue_recognize(
    p_invoice_ids => ARRAY[1001, 1002]
);

-- Amend invoice 1001's line items in the source table,
-- then re-reconcile — only correcting deltas are inserted:
SELECT * FROM ops.ops_revenue_recognize(
    p_invoice_ids => ARRAY[1001]
);

Simple event (input only)¶

A simple event is a special case — it provides only input and inserts directly, without diffing. This is what happens when you omit desired, existing, and diff_keys: there are no existing entries to diff against, so every call inserts new rows.

Use this for fire-and-forget deltas where idempotency is not needed (stock adjustments, manual corrections, one-off charges):

adjust = LedgerEvent(
    name="adjust",
    input=lambda p: select(
        p("warehouse", String).label("warehouse"),
        p("sku", String).label("sku"),
        p("value", Integer).label("value"),
        p("reason", String).label("reason"),
    ),
)

Tip

If your simple event maps one-to-one with an external identifier (e.g. an invoice_id), prefer diff mode instead. Diffing over that identifier makes the operation idempotent — calling the same event twice with the same input produces zero new rows the second time.

Factory integration¶

The recommended style is declarative: define the ledger model as a CodegenDatabaseBase subclass, then attach events with CodegenDatabaseFunctionMixin and ledger_event_function():

from sqlalchemy import Column, Integer, String
from codegen_database import CodegenDatabaseFunctionMixin
from codegen_database.ext.ledger import (
    LedgerEvent,
    construct_ledger_balances_query,
    ledger_event_function,
)
from codegen_database.factory import CodegenDatabaseLedger


class Revenue(Base):
    __tablename__ = "revenue"
    __table_args__ = {"schema": "ops"}
    __factory__ = CodegenDatabaseLedger

    invoice_id = Column(Integer, nullable=False)
    account = Column(String, nullable=False)


class RecognizeRevenue(CodegenDatabaseFunctionMixin, Base):
    __table_args__ = {"schema": "ops"}
    __funcspec__ = ledger_event_function(
        Revenue,
        LedgerEvent(
            name="recognize",
            input=lambda p: select(
                func.unnest(p("invoice_ids", ARRAY(Integer)))
                .label("invoice_id"),
            ),
            desired=lambda pginput: select(
                invoice_lines.c.invoice_id,
                invoice_lines.c.account,
                invoice_lines.c.amount.label("value"),
            ).where(
                invoice_lines.c.invoice_id.in_(
                    select(pginput.c.invoice_id)
                )
            ),
            existing=construct_ledger_balances_query("invoice_id", "account"),
            diff_keys=["invoice_id", "account"],
        ),
    )

Imperative alternative¶

If you build your schema programmatically, call CodegenDatabaseLedger and CodegenDatabaseFunction directly. Because CodegenDatabaseFunctionSpec does not include a schema, expand the spec and pass schema explicitly:

from codegen_database.factory import CodegenDatabaseLedger
from codegen_database.functions import CodegenDatabaseFunction
from codegen_database.ext.ledger import ledger_event_function

revenue = CodegenDatabaseLedger(
    tablename="revenue",
    schemaname="ops",
    metadata=metadata,
    schema_items=[
        Column("invoice_id", Integer, nullable=False),
        Column("account", String, nullable=False),
    ],
)

spec = ledger_event_function(revenue, recognize)
CodegenDatabaseFunction(
    name=spec["name"],
    schema=revenue.ctx.schemaname,
    definition=spec["definition"],
    metadata=metadata,
    returns=spec["returns"],
    language=spec["language"],
    parameters=spec["parameters"],
    security=spec["security"],
)

Naming convention¶

Generated function names follow the pattern {schema}_{table}_{name} and are registered in the ledger’s schema:

Function	Example
`{table}_{name}`	`ops_revenue_recognize`
`{table}_{name}`	`ops_inventory_adjust`

Event names must be unique within a ledger.

Validation¶

Event configuration is validated when ledger_event_function() is called:

diff_keys is required when desired is set.
existing requires desired to be set.
Every key in diff_keys must name a column on the ledger table.

Violations raise CodegenDatabaseValidationError.

codegen_database

Navigation

Related Topics