pg_accumulatorSQLAlchemy

GitHub Code Examples

Interactive demo of sqlalchemy-accumulator — the type-safe Python adapter for pg_accumulator

Registered registers: {% for r in registers %} {{ r.name }} ({{ r.kind }}){% if not loop.last %}, {% endif %} {% endfor %}
{% if ledger_sound %} ✓ Debit ≡ Credit audit passed {% else %} ⚠ Warning! Debit/Credit mismatch {% endif %}
{% with messages = get_flashed_messages(with_categories=true) %} {% if messages %} {% for category, message in messages %}
{{ message }}
{% endfor %} {% endif %} {% endwith %} {% if error %}
{{ error }}
{% endif %}
Operations
Bookkeeping (Ledger)
Orders (ORM+Accum)
Catalog (ORM)
Live Data
Query API
Code Examples
Registers

Post Movement

Record inventory receipt or shipment via handle.post()

Unpost Document

Cancel all movements for a recorder via handle.unpost()

Repost (Correct)

Atomically replace movements via handle.repost()

Create a Ledger Entry

Record double-entry posting to the General Ledger via handle.post()

Cancel a Ledger Posting

Reverses a posted ledger document and restores account balances via handle.unpost()

Account Type Guide:

  • Active (1xx/2xx/9xx) - Increases on Debit, decreases on Credit.
  • Passive (4xx/5xx/7xx) - Increases on Credit, decreases on Debit.

Trial Balance

Consolidated ledger account balances

{% if ledger_balances %} {% set total_dr = [0.0] %} {% set total_cr = [0.0] %} {% set active_total = [0.0] %} {% set passive_total = [0.0] %} {% for b in ledger_balances %} {% if total_dr.append(total_dr.pop() + (b.get('amount_dr', 0) | float)) %}{% endif %} {% if total_cr.append(total_cr.pop() + (b.get('amount_cr', 0) | float)) %}{% endif %} {% if b.get('acc_type', '') == 'A' %} {% if active_total.append(active_total.pop() + (b.get('balance', 0) | float)) %}{% endif %} {% else %} {% if passive_total.append(passive_total.pop() + (b.get('balance', 0) | float)) %}{% endif %} {% endif %} {% endfor %}
Account Type Subconto Details Debit (DR) Credit (CR) Balance Currency
{{ b.get('account', '') }} {% if b.get('acc_type', '') == 'A' %} Active {% else %} Passive {% endif %} {{ b.get('subconto', '') }} {{ "%.2f"|format(b.get('amount_dr', 0) | float) }} {{ "%.2f"|format(b.get('amount_cr', 0) | float) }} {{ "%.2f"|format(b.get('balance', 0) | float) }} {{ b.get('currency', '') }}
Total transaction sums: {{ "%.2f"|format(total_dr[0]) }} {{ "%.2f"|format(total_cr[0]) }} {% if "%.2f"|format(total_dr[0]) == "%.2f"|format(total_cr[0]) %} ✓ Totals match {% else %} ⚠ Mismatch detected! {% endif %}
Asset & Liability totals: Total assets: {{ "%.2f"|format(active_total[0]) }} USD Total liabilities & equity: {{ "%.2f"|format(passive_total[0]) }} USD
{% else %}
No bookkeeping data yet. Post first ledger entry.
{% endif %}

General Journal

Recent ledger entries (double-entry movements)

{% if ledger_movements %} {% for lm in ledger_movements %} {% endfor %}
Document Date Debit Account Debit Subconto Credit Account Credit Subconto Amount Currency Recorded At
{{ lm.get('recorder', '') }} {{ lm.get('period', '') | string | truncate(10, True, '') }} Dr{{ lm.get('account_dr', '') }} {{ lm.get('subconto_dr', '') }} Cr{{ lm.get('account_cr', '') }} {{ lm.get('subconto_cr', '') }} {{ "%.2f"|format(lm.get('amount', 0) | float) }} {{ lm.get('currency', '') }} {{ lm.get('recorded_at', '') | string | truncate(19, True, '') }}
{% else %}
General journal is empty.
{% endif %}

Create Order (ORM + Accumulator)

Creates an ORM Order and posts inventory movement in a single transaction

# What happens behind the scenes: with Session(engine) as session: # 1) ORM: create order + order line session.add(order) session.flush() # 2) Accumulator: post movement (same tx!) accum = AccumulatorClient(session) accum.use(inventory).post({...}) session.commit() # atomic!

Recent Orders

{% if orders %} {% for o in orders %} {% endfor %}
#ClientWarehouseStatusAction
{{ o.id }} {{ o.client.name if o.client else '?' }} {{ o.warehouse.name if o.warehouse else '?' }} {% if o.status == 'posted' %} ● posted {% elif o.status == 'cancelled' %} ● cancelled {% else %} ● {{ o.status }} {% endif %} {% if o.status == 'posted' %}
{% endif %}
{% else %}
No orders yet. Create one to see ORM + Accumulator in action.
{% endif %}

Warehouses (SQLAlchemy ORM)

{% if warehouses %} {% for w in warehouses %} {% endfor %}
IDNameAddress
{{ w.id }}{{ w.name }}{{ w.address or '—' }}
{% endif %}

Add Warehouse

Products (SQLAlchemy ORM)

{% if products %} {% for p in products %} {% endfor %}
IDSKUNamePriceCategory
{{ p.id }} {{ p.sku }} {{ p.name }} ${{ p.unit_price }} {{ p.category or '—' }}
{% endif %}

Add Product

Clients (SQLAlchemy ORM)

{% if clients %} {% for c in clients %} {% endfor %}
IDNameEmailPhone
{{ c.id }} {{ c.name }} {{ c.email or '—' }} {{ c.phone or '—' }}
{% endif %}

Add Client

Current Balances (from balance_cache — O(1) reads)

{% if balances %} {% for b in balances %} {% endfor %}
WarehouseProductQuantityAmount
{{ b.warehouse }} — {{ b.warehouse_name }} {{ b.product }} — {{ b.product_name }} {{ b.quantity }} {{ b.amount }}
{% else %}
No balance data yet. Post some movements first.
{% endif %}

Recent Movements (via handle.movements(limit=50))

{% if movements %} {% for m in movements %} {% endfor %}
RecorderPeriodWarehouseProductQuantityAmount
{{ m.get('recorder', '') }} {{ m.get('period', '')[:10] if m.get('period') else '' }} {{ m.get('warehouse', '') }} — {{ m.get('warehouse_name', '') }} {{ m.get('product', '') }} — {{ m.get('product_name', '') }} {{ m.get('quantity', '') }} {{ m.get('amount', '') }}
{% else %}
No movements recorded yet.
{% endif %}

Balance Query

Get instant balance via handle.balance()


            


            

                
Turnover Query

                
Aggregate turnover via handle.turnover()

                

                    

                        
                        
                    

                    

                        
                        
                    

                    

                        
                        
                    

                    

                        
                        
                    

                    
                

                

            

        


        

            
Movements Query

            
Browse movement history via handle.movements()

            

                

                    
                    
                

                

                    
                    
                

                
            

            

        

    


    
    

        

            
1. Define a Register

            
from sqlalchemy_accumulator import define_register

inventory = define_register(
    name="inventory",
    kind="balance",
    dimensions={"warehouse": "int", "product": "int"},
    resources={"quantity": "numeric(18,4)", "amount": "numeric(18,2)"},
)

        


        

            
2. Create Client & Register

            
from sqlalchemy import create_engine
from sqlalchemy_accumulator import AccumulatorClient

engine = create_engine("postgresql://user:pass@localhost/mydb")
accum = AccumulatorClient(engine)

# Create the register in the database (one-time)
accum.create_register(inventory)

        


        

            
3. Post & Query

            
handle = accum.use(inventory)

# Post a warehouse receipt
handle.post({
    "recorder": "receipt:1",
    "period":   "2026-04-01",
    "warehouse": 1, "product": 101,
    "quantity": 500, "amount": 125000,
})

# Instant balance — O(1) from materialized cache
bal = handle.balance(warehouse=1, product=101)
# {'quantity': Decimal('500'), 'amount': Decimal('125000')}

# Historical balance at a specific date
bal = handle.balance(warehouse=1, at_date="2026-04-01")

# Undo a document — all movements reversed atomically
handle.unpost("receipt:1")

# Correct a document — atomic repost
handle.repost("receipt:1", {
    "recorder": "receipt:1",
    "period":   "2026-04-01",
    "warehouse": 1, "product": 101,
    "quantity": 600, "amount": 150000,
})

        


        

            
4. ORM + Accumulator — Same Transaction

            
from sqlalchemy.orm import Session

with Session(engine) as session:
    # 1) Create ORM entities as usual
    order = Order(client_id=1, warehouse_id=1, status="posted")
    session.add(order)
    session.flush()  # get order.id

    session.add(OrderLine(
        order_id=order.id, product_id=101,
        quantity=50, unit_price=250, amount=12500,
    ))

    # 2) Post accumulator movement in the SAME transaction
    accum = AccumulatorClient(session)
    accum.use(inventory).post({
        "recorder": f"order:{order.id}",
        "period": "2026-04-19",
        "warehouse": 1, "product": 101,
        "quantity": -50, "amount": -12500,
    })

    # 3) Both ORM and accumulator commit atomically
    session.commit()
    # If anything fails — everything rolls back!

        

    


    
    

        
Registered Accumulation Registers

        

            {% if registers %}
                {% for r in registers %}
                

                    {{ r.name }}
                    {{ r.kind }}
                    {{ r.dimensions }} dims · {{ r.resources }} resources · {{ r.movements_count }} movements
                

                {% endfor %}
            {% else %}
            
No registers found. Check database connection.

            {% endif %}
        


        

            
Register Details