pg_accumulatorSQLAlchemy

GitHub Code Examples

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

{% with messages = get_flashed_messages(with_categories=true) %} {% if messages %} {% for category, message in messages %}
{{ message }}
{% endfor %} {% endif %} {% endwith %} {% if error %}
{{ error }}
{% endif %}
Operations
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 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