---
title: "Warehouse Pick Lists"
description: "Warehouse pick-list REST API: units, pick zones, waves and tote labels as incremental computed columns over Json order lines"
url: "https://www.pixeltable.com/recipes/warehouse-picklists"
source: "https://github.com/pierrebrunelle/pixeltable-warehouse-picklists"
---

# Warehouse Pick Lists

Warehouse pick-list REST API: units, pick zones, waves and tote labels as incremental computed columns over Json order lines

One `app.py` with 1 table, 5 computed columns, 6 HTTP routes. Needs `pixeltable[serve]>=0.7.15`.

## Tables

### orders (table)

| Column | Kind | Type or expression |
| --- | --- | --- |
| order_id | primary key | `` |
| customer | stored | String |
| lines | stored | Json |
| rush | stored | Bool (optional) |
| status | stored | String |
| notes | stored | Json (optional) |
| units | computed on insert | `total_units(lines)` |
| zone_list | computed on insert | `zones(lines)` |
| est_minutes | computed on insert | `pick_minutes(units, zone_list, rush)` |
| wave_name | computed on insert | `wave(zone_list, rush)` |
| tote_label | computed on insert | `label_text(order_id, wave_name, zone_list)` |

- B-tree index on `status`
- B-tree index on `wave_name`
- B-tree index on `customer`

## Routes (service `picking_api`)

- `POST /orders`: inserts into `orders` (takes order_id, customer, lines, rush, status, notes; returns order_id, units, zone_list, est_minutes, wave_name, tote_label)
- `POST /orders/update`: updates `orders` (takes status, lines, rush; returns order_id, status, units, wave_name, est_minutes)
- `POST /orders/cancel`: deletes from `orders`
- `GET /waves/queue`: runs the query `wave_queue` (takes wave_name, status)
- `GET /orders/label`: runs the query `order_label` (takes order_id)
- `GET /customers/orders`: runs the query `customer_orders` (takes customer)

## Run it locally

```sh
git clone https://github.com/pierrebrunelle/pixeltable-warehouse-picklists
cd pixeltable-warehouse-picklists
uv sync
uv run pxt schema update app.py picking
uv run python seed.py picking
uv run pxt service run app.py picking --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 `8f37e6f1` (https://github.com/pierrebrunelle/pixeltable-warehouse-picklists/blob/8f37e6f15c7914c89ff64cd0e0f0e7b7fbb830f3/app.py):

```python
"""Warehouse Pick-List API built with Pixeltable.

Orders arrive with a Json list of lines. Total units, pick zones, estimated pick time, pick wave and the
tote label are computed columns (Python UDFs in udfs.py) that Pixeltable keeps up to date on every insert
and update. A FastAPIRouter serves insert / update / delete routes plus index-backed queries for the
pick floor (open orders in a wave, one order's tote label).

Layout: UDFs live in udfs.py (imported below); this file holds the tables, the @pxt.query functions and the routes.

Run locally:
    pxt schema update app.py picking
    pxt service run app.py picking

See README.md for the walkthrough and Pixeltable Cloud deployment.
"""
import pixeltable as pxt
from pixeltable.serving import FastAPIRouter

from udfs import label_text, pick_minutes, total_units, wave, zones

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


class Orders(TableModel, name='orders', has_default_idxs=False):
    order_id = pxt.Column(type=pxt.String, primary_key=True)
    customer: pxt.String
    lines: pxt.Json
    rush: pxt.Bool | None
    status: pxt.String  # open -> picking -> packed -> shipped
    notes: pxt.Json | None

    # computed columns: chained UDFs, recomputed when an input column changes
    units = total_units(lines)
    zone_list = zones(lines)
    est_minutes = pick_minutes(units, zone_list, rush)
    wave_name = wave(zone_list, rush)
    tote_label = label_text(order_id, wave_name, zone_list)

    __indexes__ = [pxt.BtreeIndex(status), pxt.BtreeIndex(wave_name), pxt.BtreeIndex(customer)]


# ---- queries ----
@pxt.query
def wave_queue(wave_name: str, status: str):
    """Orders in a pick wave with a given status, longest pick first (index-backed)."""
    return (
        Orders.where((Orders.wave_name == wave_name) & (Orders.status == status))
        .select(Orders.order_id, Orders.customer, Orders.units, Orders.est_minutes, Orders.tote_label)
        .order_by(Orders.est_minutes, asc=False)
    )


@pxt.query
def order_label(order_id: str):
    """One order's tote label and pick estimate (0 rows -> 404)."""
    return Orders.where(Orders.order_id == order_id).select(
        Orders.order_id, Orders.tote_label, Orders.est_minutes, Orders.status
    )


@pxt.query
def customer_orders(customer: str):
    """A customer's orders, newest id first."""
    return (
        Orders.where(Orders.customer == customer)
        .select(Orders.order_id, Orders.status, Orders.units, Orders.wave_name)
        .order_by(Orders.order_id, asc=False)
    )


# ---- routes ----
picking_api = FastAPIRouter(name='picking_api')

picking_api.add_insert_route(
    Orders,
    path='/orders',
    inputs=[Orders.order_id, Orders.customer, Orders.lines, Orders.rush, Orders.status, Orders.notes],
    outputs=[Orders.order_id, Orders.units, Orders.zone_list, Orders.est_minutes, Orders.wave_name, Orders.tote_label],
)
picking_api.add_update_route(
    Orders,
    path='/orders/update',
    inputs=[Orders.status, Orders.lines, Orders.rush],
    outputs=[Orders.order_id, Orders.status, Orders.units, Orders.wave_name, Orders.est_minutes],
)
picking_api.add_delete_route(Orders, path='/orders/cancel')
picking_api.add_query_route(path='/waves/queue', query=wave_queue, method='get')
picking_api.add_query_route(path='/orders/label', query=order_label, one_row=True, method='get')
picking_api.add_query_route(path='/customers/orders', query=customer_orders, method='get')
```