Invoice Audit

1"""Invoice Audit API built with Pixeltable.
2
3    pxt schema update app.py audit
4    pxt service run app.py audit
5"""
6import pixeltable as pxt
7import pixeltable.functions as pxtf
8from pixeltable.serving import FastAPIRouter
9
10from udfs import (
11    apply_discount,
12    excerpt_note,
13    grand_total,
14    invoice_ref,
15    line_subtotal,
16    risk_band,
17    risk_score,
18    tax_amount,
19)
20
21# ---- tables ----
22TableModel = pxt.model_base()
23
24
25class Vendors(TableModel, name='vendors'):
26    id = pxt.Column(value=pxtf.uuid.uuid7(), primary_key=True)
27    code: pxt.String
28    name: pxt.String
29    tier: pxt.String
30    active: pxt.Bool
31
32    code_upper = pxtf.string.upper(code)
33    tier_upper = pxtf.string.upper(tier)
34
35
36class LineItems(TableModel, name='line_items'):
37    id = pxt.Column(value=pxtf.uuid.uuid7(), primary_key=True)
38    vendor_code: pxt.String
39    sku: pxt.String
40    qty: pxt.Int
41    unit_price: pxt.Float
42    discount_pct: pxt.Float
43    tax_rate: pxt.Float
44    vendor_tier: pxt.String
45    note: pxt.String | None
46    auditor: pxt.String | None
47
48    ref = invoice_ref(vendor_code, sku)
49    subtotal = line_subtotal(qty, unit_price)
50    net = apply_discount(subtotal, discount_pct)
51    tax = tax_amount(net, tax_rate)
52    total = grand_total(net, tax)
53    risk = risk_score(qty, unit_price, discount_pct, vendor_tier)
54    band = risk_band(risk)
55    note_excerpt = excerpt_note(note, n=48)
56
57
58# ---- queries ----
59@pxt.query
60def flagged(vendor_code: str):
61    """Lines for a vendor that are not 'clear', riskiest first."""
62    return LineItems.where((LineItems.vendor_code == vendor_code) & (LineItems.band != 'clear')).select(
63        LineItems.ref, LineItems.total, LineItems.risk, LineItems.band, LineItems.note_excerpt
64    ).order_by(LineItems.risk, asc=False)
65
66
67# ---- routes ----
68audit = FastAPIRouter(name='audit')
69audit.add_insert_route(
70    LineItems, path='/lines',
71    inputs=[LineItems.vendor_code, LineItems.sku, LineItems.qty, LineItems.unit_price, LineItems.discount_pct,
72            LineItems.tax_rate, LineItems.vendor_tier, LineItems.note, LineItems.auditor],
73    outputs=[LineItems.id, LineItems.ref, LineItems.subtotal, LineItems.net, LineItems.tax, LineItems.total,
74             LineItems.risk, LineItems.band, LineItems.note_excerpt],
75)
76audit.add_update_route(LineItems, path='/lines/adjust', inputs=[LineItems.discount_pct, LineItems.auditor],
77                       outputs=[LineItems.id, LineItems.net, LineItems.total, LineItems.risk, LineItems.band])
78audit.add_insert_route(Vendors, path='/vendors', inputs=[Vendors.code, Vendors.name, Vendors.tier, Vendors.active],
79                       outputs=[Vendors.id, Vendors.code_upper, Vendors.tier_upper])
80audit.add_compute_route(LineItems, path='/risk',
81                        inputs=[LineItems.qty, LineItems.unit_price, LineItems.discount_pct, LineItems.vendor_tier],
82                        outputs=[LineItems.risk, LineItems.band])
83audit.add_query_route(path='/flagged', query=flagged, method='get')