---
title: "Invoice Audit"
description: "Invoice line-item audit API: a chain of computed columns from subtotal to tax, total and risk band"
url: "https://www.pixeltable.com/recipes/invoice-audit"
source: "https://github.com/pierrebrunelle/pixeltable-invoice-audit"
---

# Invoice Audit

Invoice line-item audit API: a chain of computed columns from subtotal to tax, total and risk band

One `app.py` with 2 tables, 10 computed columns, 5 HTTP routes. Needs `pixeltable[serve]>=0.7.14`.

## Tables

### vendors (table)

| Column | Kind | Type or expression |
| --- | --- | --- |
| id | primary key | `pxtf.uuid.uuid7()` |
| code | stored | String |
| name | stored | String |
| tier | stored | String |
| active | stored | Bool |
| code_upper | computed on insert | `pxtf.string.upper(code)` |
| tier_upper | computed on insert | `pxtf.string.upper(tier)` |

### line_items (table)

| Column | Kind | Type or expression |
| --- | --- | --- |
| id | primary key | `pxtf.uuid.uuid7()` |
| vendor_code | stored | String |
| sku | stored | String |
| qty | stored | Int |
| unit_price | stored | Float |
| discount_pct | stored | Float |
| tax_rate | stored | Float |
| vendor_tier | stored | String |
| note | stored | String (optional) |
| auditor | stored | String (optional) |
| ref | computed on insert | `invoice_ref(vendor_code, sku)` |
| subtotal | computed on insert | `line_subtotal(qty, unit_price)` |
| net | computed on insert | `apply_discount(subtotal, discount_pct)` |
| tax | computed on insert | `tax_amount(net, tax_rate)` |
| total | computed on insert | `grand_total(net, tax)` |
| risk | computed on insert | `risk_score(qty, unit_price, discount_pct, vendor_tier)` |
| band | computed on insert | `risk_band(risk)` |
| note_excerpt | computed on insert | `excerpt_note(note, n=48)` |

## Routes (service `audit`)

- `POST /lines`: inserts into `line_items` (takes vendor_code, sku, qty, unit_price, discount_pct, tax_rate, vendor_tier, note, auditor; returns id, ref, subtotal, net, tax, total, risk, band, note_excerpt)
- `POST /lines/adjust`: updates `line_items` (takes discount_pct, auditor; returns id, net, total, risk, band)
- `POST /vendors`: inserts into `vendors` (takes code, name, tier, active; returns id, code_upper, tier_upper)
- `POST /risk`: computes `line_items` columns without storing (takes qty, unit_price, discount_pct, vendor_tier; returns risk, band)
- `GET /flagged`: runs the query `flagged` (takes vendor_code)

## Run it locally

```sh
git clone https://github.com/pierrebrunelle/pixeltable-invoice-audit
cd pixeltable-invoice-audit
uv sync
uv run pxt schema update app.py audit
uv run python seed.py audit
uv run pxt service run app.py audit --port 8000
```

## Run it on Pixeltable Cloud

```sh
uv run pxt new    # a free database, no account
# put the pxt:// address it prints in pixeltable.toml
uv run pxt db update pxt://<org>:main -f
uv run pxt schema update app.py pxt://<org>:main -f
uv run pxt service update app.py pxt://<org>:main -f
```

## app.py

At commit `b3b2faa4` (https://github.com/pierrebrunelle/pixeltable-invoice-audit/blob/b3b2faa4436741a156bb0e3b7821b8956cc1786f/app.py):

```python
"""Invoice Audit API built with Pixeltable.

    pxt schema update app.py audit
    pxt service run app.py audit
"""
import pixeltable as pxt
import pixeltable.functions as pxtf
from pixeltable.serving import FastAPIRouter

from udfs import (
    apply_discount,
    excerpt_note,
    grand_total,
    invoice_ref,
    line_subtotal,
    risk_band,
    risk_score,
    tax_amount,
)

# ---- tables ----
TableModel = pxt.model_base()


class Vendors(TableModel, name='vendors'):
    id = pxt.Column(value=pxtf.uuid.uuid7(), primary_key=True)
    code: pxt.String
    name: pxt.String
    tier: pxt.String
    active: pxt.Bool

    code_upper = pxtf.string.upper(code)
    tier_upper = pxtf.string.upper(tier)


class LineItems(TableModel, name='line_items'):
    id = pxt.Column(value=pxtf.uuid.uuid7(), primary_key=True)
    vendor_code: pxt.String
    sku: pxt.String
    qty: pxt.Int
    unit_price: pxt.Float
    discount_pct: pxt.Float
    tax_rate: pxt.Float
    vendor_tier: pxt.String
    note: pxt.String | None
    auditor: pxt.String | None

    ref = invoice_ref(vendor_code, sku)
    subtotal = line_subtotal(qty, unit_price)
    net = apply_discount(subtotal, discount_pct)
    tax = tax_amount(net, tax_rate)
    total = grand_total(net, tax)
    risk = risk_score(qty, unit_price, discount_pct, vendor_tier)
    band = risk_band(risk)
    note_excerpt = excerpt_note(note, n=48)


# ---- queries ----
@pxt.query
def flagged(vendor_code: str):
    """Lines for a vendor that are not 'clear', riskiest first."""
    return LineItems.where((LineItems.vendor_code == vendor_code) & (LineItems.band != 'clear')).select(
        LineItems.ref, LineItems.total, LineItems.risk, LineItems.band, LineItems.note_excerpt
    ).order_by(LineItems.risk, asc=False)


# ---- routes ----
audit = FastAPIRouter(name='audit')
audit.add_insert_route(
    LineItems, path='/lines',
    inputs=[LineItems.vendor_code, LineItems.sku, LineItems.qty, LineItems.unit_price, LineItems.discount_pct,
            LineItems.tax_rate, LineItems.vendor_tier, LineItems.note, LineItems.auditor],
    outputs=[LineItems.id, LineItems.ref, LineItems.subtotal, LineItems.net, LineItems.tax, LineItems.total,
             LineItems.risk, LineItems.band, LineItems.note_excerpt],
)
audit.add_update_route(LineItems, path='/lines/adjust', inputs=[LineItems.discount_pct, LineItems.auditor],
                       outputs=[LineItems.id, LineItems.net, LineItems.total, LineItems.risk, LineItems.band])
audit.add_insert_route(Vendors, path='/vendors', inputs=[Vendors.code, Vendors.name, Vendors.tier, Vendors.active],
                       outputs=[Vendors.id, Vendors.code_upper, Vendors.tier_upper])
audit.add_compute_route(LineItems, path='/risk',
                        inputs=[LineItems.qty, LineItems.unit_price, LineItems.discount_pct, LineItems.vendor_tier],
                        outputs=[LineItems.risk, LineItems.band])
audit.add_query_route(path='/flagged', query=flagged, method='get')
```