---
title: "Lab Accession"
description: "Lab sample accession queue API: three related tables, parallel writers and racing claims"
url: "https://www.pixeltable.com/recipes/lab-accession"
source: "https://github.com/pierrebrunelle/pixeltable-lab-accession"
---

# Lab Accession

Lab sample accession queue API: three related tables, parallel writers and racing claims

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

## Tables

### stations (table)

| Column | Kind | Type or expression |
| --- | --- | --- |
| station_id | primary key | `` |
| name | stored | String |
| capacity | stored | Int |
| label | computed on insert | `station_label(name, capacity)` |

### samples (table)

| Column | Kind | Type or expression |
| --- | --- | --- |
| accession_id | primary key | `` |
| specimen | stored | String |
| priority | stored | Int |
| received_at | stored | String |
| batch_code | stored | String (optional) |
| priority_label | computed on insert | `priority_name(priority)` |

### assignments (table)

| Column | Kind | Type or expression |
| --- | --- | --- |
| id | primary key | `pxtf.uuid.uuid7()` |
| accession_id | stored | String |
| station_id | stored | String |
| priority | stored | Int |
| delayed_min | stored | Int |
| status | stored | String |
| claimed_by | stored | String (optional) |
| urgency | computed on insert | `urgency_band(priority, delayed_min)` |

- B-tree index on `station_id`
- B-tree index on `status`

## Routes (service `lab_api`)

- `POST /stations`: inserts into `stations` (takes station_id, name, capacity; returns station_id, label)
- `POST /samples`: inserts into `samples` (takes accession_id, specimen, priority, received_at, batch_code; returns accession_id, priority_label)
- `POST /assignments`: inserts into `assignments` (takes accession_id, station_id, priority, delayed_min, status, claimed_by; returns id, urgency)
- `POST /assignments/claim`: updates `assignments` (takes status, claimed_by; returns id, status, claimed_by)
- `POST /urgency`: computes `assignments` columns without storing (takes priority, delayed_min; returns urgency)
- `GET /stations/queue`: runs the query `station_queue` (takes station_id)
- `GET /batches`: runs the query `batch` (takes batch_code)

## Run it locally

```sh
git clone https://github.com/pierrebrunelle/pixeltable-lab-accession
cd pixeltable-lab-accession
uv sync
uv run pxt schema update app.py lab
uv run python seed.py lab
uv run pxt service run app.py lab --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 `d21fd450` (https://github.com/pierrebrunelle/pixeltable-lab-accession/blob/d21fd450237f76a69efc07901de94a67e0e8a1a1/app.py):

```python
"""Lab Accession API built with Pixeltable.

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

from udfs import priority_name, station_label, urgency_band

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


class Stations(TableModel, name='stations'):
    station_id = pxt.Column(type=pxt.String, primary_key=True)
    name: pxt.String
    capacity: pxt.Int

    label = station_label(name, capacity)


class Samples(TableModel, name='samples'):
    accession_id = pxt.Column(type=pxt.String, primary_key=True)
    specimen: pxt.String
    priority: pxt.Int
    received_at: pxt.String
    batch_code: pxt.String | None

    priority_label = priority_name(priority)


class Assignments(TableModel, name='assignments', has_default_idxs=False):
    id = pxt.Column(value=pxtf.uuid.uuid7(), primary_key=True)
    accession_id: pxt.String
    station_id: pxt.String
    priority: pxt.Int
    delayed_min: pxt.Int
    status: pxt.String
    claimed_by: pxt.String | None

    urgency = urgency_band(priority, delayed_min)

    __indexes__ = [pxt.BtreeIndex(station_id), pxt.BtreeIndex(status)]


# ---- queries ----
@pxt.query
def station_queue(station_id: str):
    """Work for one station, most urgent first."""
    return Assignments.where(Assignments.station_id == station_id).select(
        Assignments.id, Assignments.accession_id, Assignments.urgency, Assignments.status, Assignments.claimed_by
    ).order_by(Assignments.priority, asc=False)


@pxt.query
def batch(batch_code: str):
    """Samples received in one batch."""
    return Samples.where(Samples.batch_code == batch_code).select(
        Samples.accession_id, Samples.specimen, Samples.priority_label
    ).order_by(Samples.accession_id)


# ---- routes ----
lab_api = FastAPIRouter(name='lab_api')
lab_api.add_insert_route(Stations, path='/stations', inputs=[Stations.station_id, Stations.name, Stations.capacity],
                         outputs=[Stations.station_id, Stations.label])
lab_api.add_insert_route(Samples, path='/samples',
                         inputs=[Samples.accession_id, Samples.specimen, Samples.priority, Samples.received_at,
                                 Samples.batch_code],
                         outputs=[Samples.accession_id, Samples.priority_label])
lab_api.add_insert_route(Assignments, path='/assignments',
                         inputs=[Assignments.accession_id, Assignments.station_id, Assignments.priority,
                                 Assignments.delayed_min, Assignments.status, Assignments.claimed_by],
                         outputs=[Assignments.id, Assignments.urgency])
lab_api.add_update_route(Assignments, path='/assignments/claim', inputs=[Assignments.status, Assignments.claimed_by],
                         outputs=[Assignments.id, Assignments.status, Assignments.claimed_by])
lab_api.add_compute_route(Assignments, path='/urgency', inputs=[Assignments.priority, Assignments.delayed_min],
                          outputs=[Assignments.urgency])
lab_api.add_query_route(path='/stations/queue', query=station_queue, method='get')
lab_api.add_query_route(path='/batches', query=batch, method='get')
```