# Recipes

> A small client with pagination and 429 backoff, and runnable snippets for common jobs.

Every recipe imports the client below. Save it as `cubic.mts` or `cubic.py`. Both read the key from `CUBIC_API_KEY`.

- TypeScript: no dependencies, Node 18+. Run with `npx tsx <file>.mts` or `bun <file>.mts`.
- Python: `pip install requests`, Python 3.9+. Run with `python <file>.py`.

TypeScript types: `npx openapi-typescript https://developers.gocubic.io/openapi.json -o cubic-api.d.ts`.

## Client

Follows cursors, waits out `429` using `Retry-After`, retries `5xx` with backoff, and raises everything else with the API's `code` and `hint`.

```ts
// cubic.mts
const BASE_URL = 'https://api.gocubic.io/v1';
const API_KEY = process.env.CUBIC_API_KEY;
if (!API_KEY) throw new Error('Set the CUBIC_API_KEY environment variable');

export interface MilestoneDate {
  date: string;
  time: string | null;
  actual: boolean;
}

export interface Place {
  name: string | null;
  code: string | null;
  countryCode: string | null;
}

export interface Milestone {
  type: string;
  place: Place | null;
  arrival: MilestoneDate | null;
  departure: MilestoneDate | null;
  completed: boolean;
}

export interface Shipment {
  reference: string;
  status: string;
  serviceType: string;
  tags: string[];
  portOfDischarge: Place | null;
  milestones: Milestone[];
  currentMilestoneIndex: number;
  updatedAt: string;
  [field: string]: unknown;
}

export type Query = Record<string, string | number | string[] | undefined>;

export class CubicApiError extends Error {
  problem: { status: number; code: string; detail: string; hint: string; parameter?: string };

  constructor(problem: CubicApiError['problem']) {
    super(`${problem.status} ${problem.code}: ${problem.detail} Hint: ${problem.hint}`);
    this.problem = problem;
  }
}

const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));

export async function cubicGet<T>(path: string, query: Query = {}): Promise<T> {
  const url = new URL(BASE_URL + path);
  for (const [key, value] of Object.entries(query)) {
    if (value === undefined) continue;
    for (const item of Array.isArray(value) ? value : [value]) url.searchParams.append(key, String(item));
  }

  for (let attempt = 0; ; attempt++) {
    const response = await fetch(url, { headers: { Authorization: `Bearer ${API_KEY}` } });
    if (response.ok) return (await response.json()) as T;

    const retryable = response.status === 429 || response.status >= 500;
    if (retryable && attempt < 5) {
      const retryAfter = Number(response.headers.get('Retry-After'));
      await sleep((retryAfter > 0 ? retryAfter : 2 ** attempt) * 1000);
      continue;
    }

    const problem = await response
      .json()
      .catch(() => ({ status: response.status, code: 'unknown', detail: response.statusText, hint: '' }));
    throw new CubicApiError(problem);
  }
}

export async function* listShipments(query: Query = {}): AsyncGenerator<Shipment> {
  let cursor: string | undefined;
  do {
    const page = await cubicGet<{ data: Shipment[]; nextCursor: string | null }>('/shipments', {
      limit: 100,
      ...query,
      cursor,
    });
    yield* page.data;
    cursor = page.nextCursor ?? undefined;
  } while (cursor);
}

export const findMilestone = (shipment: Shipment, type: string) => shipment.milestones.find((milestone) => milestone.type === type) ?? null;
```

```python
# cubic.py
import os
import time

import requests

BASE_URL = "https://api.gocubic.io/v1"

session = requests.Session()
session.headers["Authorization"] = f"Bearer {os.environ['CUBIC_API_KEY']}"


class CubicApiError(Exception):
    def __init__(self, problem):
        super().__init__(
            f"{problem.get('status')} {problem.get('code')}: {problem.get('detail')} Hint: {problem.get('hint')}"
        )
        self.problem = problem


def cubic_get(path, params=None):
    for attempt in range(6):
        response = session.get(BASE_URL + path, params=params, timeout=30)
        if response.ok:
            return response.json()

        retryable = response.status_code == 429 or response.status_code >= 500
        if retryable and attempt < 5:
            retry_after = response.headers.get("Retry-After")
            time.sleep(int(retry_after) if retry_after else 2**attempt)
            continue

        try:
            problem = response.json()
        except ValueError:
            problem = {"status": response.status_code, "code": "unknown", "detail": response.reason, "hint": ""}
        raise CubicApiError(problem)


def list_shipments(**filters):
    """Yield every shipment matching the filters, following cursors. A list value repeats the parameter."""
    params = {"limit": 100, **filters}
    while True:
        page = cubic_get("/shipments", params)
        yield from page["data"]
        if page["nextCursor"] is None:
            return
        params["cursor"] = page["nextCursor"]


def find_milestone(shipment, milestone_type):
    return next((milestone for milestone in shipment["milestones"] if milestone["type"] == milestone_type), None)
```

## Verify the connection

```ts
import { cubicGet } from './cubic.mts';

const me = await cubicGet<{ account: { name: string }; apiKey: { name: string; prefix: string } }>('/me');
console.log(`Connected to "${me.account.name}" using key "${me.apiKey.name}" (${me.apiKey.prefix}…)`);
```

```python
from cubic import cubic_get

me = cubic_get("/me")
print(f"Connected to \"{me['account']['name']}\" using key \"{me['apiKey']['name']}\" ({me['apiKey']['prefix']}…)")
```

## List active shipments with ETA

The ETA is a milestone's `arrival` while `actual` is `false`. Either milestone, or its `arrival`, can be missing. `currentMilestoneIndex` says where the shipment is now.

```ts
import { findMilestone, listShipments, type MilestoneDate } from './cubic.mts';

const describe = (arrival: MilestoneDate | null | undefined) =>
  arrival ? `${arrival.date} (${arrival.actual ? 'actual' : 'estimated'})` : 'not known yet';

for await (const shipment of listShipments({ status: 'Active' })) {
  const port = findMilestone(shipment, 'DestinationPort');
  const delivery = findMilestone(shipment, 'Delivery');
  console.log(
    `${shipment.reference}  ${shipment.serviceType}  ` +
      `current milestone ${shipment.milestones[shipment.currentMilestoneIndex].type}  ` +
      `port of discharge ${shipment.portOfDischarge?.name ?? '—'}: ${describe(port?.arrival)}  ` +
      `delivery: ${describe(delivery?.arrival)}`,
  );
}
```

```python
from cubic import find_milestone, list_shipments


def describe(milestone):
    arrival = milestone and milestone["arrival"]
    if not arrival:
        return "not known yet"
    return f"{arrival['date']} ({'actual' if arrival['actual'] else 'estimated'})"


for shipment in list_shipments(status="Active"):
    port = find_milestone(shipment, "DestinationPort")
    delivery = find_milestone(shipment, "Delivery")
    port_name = (shipment["portOfDischarge"] or {}).get("name") or "—"
    print(
        f"{shipment['reference']}  {shipment['serviceType']}  "
        f"current milestone {shipment['milestones'][shipment['currentMilestoneIndex']]['type']}  "
        f"port of discharge {port_name}: {describe(port)}  delivery: {describe(delivery)}"
    )
```

## Find a shipment by PO, container, master bill or FBA ID

`tag` is an exact match on the customer's own tags (PO numbers). `search` is free text across the reference, tags, master bill / air waybill numbers, container numbers, and Amazon FBA and reference IDs.

```ts
import { listShipments } from './cubic.mts';

const [kind, value] = process.argv.slice(2);
if (!value || (kind !== 'tag' && kind !== 'search')) {
  throw new Error('Usage: find.mts tag PO-8841 | find.mts search CSNU1234567');
}

let found = 0;
for await (const shipment of listShipments({ [kind]: value })) {
  found++;
  console.log(shipment.reference, shipment.status, shipment.tags.join(','));
}
if (found === 0) console.log(`No shipment matches ${kind}=${value}`);
```

```python
import sys

from cubic import list_shipments

if len(sys.argv) != 3 or sys.argv[1] not in ("tag", "search"):
    sys.exit("Usage: find.py tag PO-8841 | find.py search CSNU1234567")
kind, value = sys.argv[1:]

matches = list(list_shipments(**{kind: value}))
for shipment in matches:
    print(shipment["reference"], shipment["status"], ",".join(shipment["tags"]))
if not matches:
    print(f"No shipment matches {kind}={value}")
```

## Poll for changes

Store the greatest `updatedAt` you have seen and pass it back as `updatedSince`. It is inclusive, so the boundary shipment comes back again: upsert by `reference`. The first run reads everything.

```ts
import { existsSync, readFileSync, writeFileSync } from 'node:fs';
import { listShipments, type Shipment } from './cubic.mts';

const STATE_FILE = 'cubic-sync-state.json';
const POLL_INTERVAL_MS = 5 * 60 * 1000;

const handleChange = (shipment: Shipment) => {
  console.log(`${shipment.reference} is ${shipment.status} (updated ${shipment.updatedAt})`);
};

let updatedSince: string | undefined = existsSync(STATE_FILE)
  ? JSON.parse(readFileSync(STATE_FILE, 'utf8')).updatedSince
  : undefined;

for (;;) {
  let newest: string | undefined;
  for await (const shipment of listShipments({ updatedSince })) {
    newest ??= shipment.updatedAt;
    handleChange(shipment);
  }
  if (newest) {
    updatedSince = newest;
    writeFileSync(STATE_FILE, JSON.stringify({ updatedSince }));
  }
  await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL_MS));
}
```

```python
import json
import os
import time

from cubic import list_shipments

STATE_FILE = "cubic-sync-state.json"
POLL_INTERVAL_SECONDS = 5 * 60


def handle_change(shipment):
    print(f"{shipment['reference']} is {shipment['status']} (updated {shipment['updatedAt']})")


updated_since = None
if os.path.exists(STATE_FILE):
    with open(STATE_FILE) as state_file:
        updated_since = json.load(state_file)["updatedSince"]

while True:
    filters = {"updatedSince": updated_since} if updated_since else {}
    newest = None
    for shipment in list_shipments(**filters):
        newest = newest or shipment["updatedAt"]
        handle_change(shipment)
    if newest:
        updated_since = newest
        with open(STATE_FILE, "w") as state_file:
            json.dump({"updatedSince": updated_since}, state_file)
    time.sleep(POLL_INTERVAL_SECONDS)
```

## What is arriving this week

Milestone dates compare correctly as text. This uses the `Delivery` milestone; use `DestinationPort` for arrivals at the port of discharge.

```ts
import { findMilestone, listShipments } from './cubic.mts';

const isoDate = (date: Date) => date.toLocaleDateString('en-CA');
const today = new Date();
const from = isoDate(today);
const to = isoDate(new Date(today.getFullYear(), today.getMonth(), today.getDate() + 7));

const arriving = [];
for await (const shipment of listShipments({ status: 'Active' })) {
  const arrival = findMilestone(shipment, 'Delivery')?.arrival;
  if (arrival && arrival.date >= from && arrival.date < to) arriving.push({ shipment, arrival });
}

arriving.sort((a, b) => a.arrival.date.localeCompare(b.arrival.date));
for (const { shipment, arrival } of arriving) {
  console.log(`${arrival.date}${arrival.actual ? '' : ' (estimated)'}  ${shipment.reference}  ${shipment.tags.join(',')}`);
}
```

```python
from datetime import date, timedelta

from cubic import find_milestone, list_shipments

start = date.today().isoformat()
end = (date.today() + timedelta(days=7)).isoformat()

arriving = []
for shipment in list_shipments(status="Active"):
    delivery = find_milestone(shipment, "Delivery")
    arrival = delivery and delivery["arrival"]
    if arrival and start <= arrival["date"] < end:
        arriving.append((arrival, shipment))

for arrival, shipment in sorted(arriving, key=lambda pair: pair[0]["date"]):
    suffix = "" if arrival["actual"] else " (estimated)"
    print(f"{arrival['date']}{suffix}  {shipment['reference']}  {','.join(shipment['tags'])}")
```

## Export to a spreadsheet

One row per shipment in `shipments.csv`. `null` becomes an empty cell.

```ts
import { writeFileSync } from 'node:fs';
import { findMilestone, listShipments, type Shipment } from './cubic.mts';

const COLUMNS: [string, (shipment: Shipment) => unknown][] = [
  ['reference', (s) => s.reference],
  ['status', (s) => s.status],
  ['serviceType', (s) => s.serviceType],
  ['tags', (s) => s.tags.join(' ')],
  ['portOfDischarge', (s) => s.portOfDischarge?.name],
  ['portArrival', (s) => findMilestone(s, 'DestinationPort')?.arrival?.date],
  ['portArrivalIsActual', (s) => findMilestone(s, 'DestinationPort')?.arrival?.actual],
  ['deliveryArrival', (s) => findMilestone(s, 'Delivery')?.arrival?.date],
  ['deliveryArrivalIsActual', (s) => findMilestone(s, 'Delivery')?.arrival?.actual],
  ['updatedAt', (s) => s.updatedAt],
];

const cell = (value: unknown) => {
  const text = value === null || value === undefined ? '' : String(value);
  return /[",\n]/.test(text) ? `"${text.replace(/"/g, '""')}"` : text;
};

const lines = [COLUMNS.map(([header]) => header).join(',')];
for await (const shipment of listShipments()) {
  lines.push(COLUMNS.map(([, read]) => cell(read(shipment))).join(','));
}

writeFileSync('shipments.csv', lines.join('\n') + '\n');
console.log(`Wrote ${lines.length - 1} shipments to shipments.csv`);
```

```python
import csv

from cubic import find_milestone, list_shipments


def arrival(shipment, milestone_type, field):
    milestone = find_milestone(shipment, milestone_type)
    return milestone["arrival"][field] if milestone and milestone["arrival"] else None


COLUMNS = [
    ("reference", lambda s: s["reference"]),
    ("status", lambda s: s["status"]),
    ("serviceType", lambda s: s["serviceType"]),
    ("tags", lambda s: " ".join(s["tags"])),
    ("portOfDischarge", lambda s: (s["portOfDischarge"] or {}).get("name")),
    ("portArrival", lambda s: arrival(s, "DestinationPort", "date")),
    ("portArrivalIsActual", lambda s: arrival(s, "DestinationPort", "actual")),
    ("deliveryArrival", lambda s: arrival(s, "Delivery", "date")),
    ("deliveryArrivalIsActual", lambda s: arrival(s, "Delivery", "actual")),
    ("updatedAt", lambda s: s["updatedAt"]),
]

count = 0
with open("shipments.csv", "w", newline="") as csv_file:
    writer = csv.writer(csv_file)
    writer.writerow([header for header, _ in COLUMNS])
    for shipment in list_shipments():
        writer.writerow([read(shipment) for _, read in COLUMNS])
        count += 1
print(f"Wrote {count} shipments to shipments.csv")
```

## Sync to a database

Incremental sync into SQLite: an upsert keyed on `reference`, resuming from the greatest `updated_at` stored. Run it on a schedule.

```ts
// Requires Node 22.13+ for node:sqlite.
import { DatabaseSync } from 'node:sqlite';
import { findMilestone, listShipments } from './cubic.mts';

const db = new DatabaseSync('cubic.db');
db.exec(`CREATE TABLE IF NOT EXISTS shipments (
  reference TEXT PRIMARY KEY,
  status TEXT NOT NULL,
  service_type TEXT NOT NULL,
  delivery_date TEXT,
  delivery_is_actual INTEGER,
  updated_at TEXT NOT NULL,
  json TEXT NOT NULL
)`);

const upsert = db.prepare(`INSERT INTO shipments VALUES (?, ?, ?, ?, ?, ?, ?)
  ON CONFLICT(reference) DO UPDATE SET
    status = excluded.status,
    service_type = excluded.service_type,
    delivery_date = excluded.delivery_date,
    delivery_is_actual = excluded.delivery_is_actual,
    updated_at = excluded.updated_at,
    json = excluded.json`);

const last = db.prepare('SELECT MAX(updated_at) AS value FROM shipments').get() as { value: string | null };

// One transaction: the list is newest first, so a partial run would move MAX(updated_at) past rows it never stored.
let count = 0;
db.exec('BEGIN');
try {
  for await (const shipment of listShipments({ updatedSince: last.value ?? undefined })) {
    const arrival = findMilestone(shipment, 'Delivery')?.arrival ?? null;
    upsert.run(
      shipment.reference,
      shipment.status,
      shipment.serviceType,
      arrival ? arrival.date : null,
      arrival ? Number(arrival.actual) : null,
      shipment.updatedAt,
      JSON.stringify(shipment),
    );
    count++;
  }
  db.exec('COMMIT');
} catch (error) {
  db.exec('ROLLBACK');
  throw error;
}
console.log(`Upserted ${count} shipments`);
```

```python
import json
import sqlite3

from cubic import find_milestone, list_shipments

db = sqlite3.connect("cubic.db")
db.execute(
    """CREATE TABLE IF NOT EXISTS shipments (
        reference TEXT PRIMARY KEY,
        status TEXT NOT NULL,
        service_type TEXT NOT NULL,
        delivery_date TEXT,
        delivery_is_actual INTEGER,
        updated_at TEXT NOT NULL,
        json TEXT NOT NULL
    )"""
)

last = db.execute("SELECT MAX(updated_at) FROM shipments").fetchone()[0]
filters = {"updatedSince": last} if last else {}

count = 0
for shipment in list_shipments(**filters):
    delivery = find_milestone(shipment, "Delivery")
    arrival = delivery and delivery["arrival"]
    db.execute(
        """INSERT INTO shipments VALUES (?, ?, ?, ?, ?, ?, ?)
           ON CONFLICT(reference) DO UPDATE SET
             status = excluded.status,
             service_type = excluded.service_type,
             delivery_date = excluded.delivery_date,
             delivery_is_actual = excluded.delivery_is_actual,
             updated_at = excluded.updated_at,
             json = excluded.json""",
        (
            shipment["reference"],
            shipment["status"],
            shipment["serviceType"],
            arrival["date"] if arrival else None,
            int(arrival["actual"]) if arrival else None,
            shipment["updatedAt"],
            json.dumps(shipment),
        ),
    )
    count += 1
db.commit()
print(f"Upserted {count} shipments")
```

## Handle 429 with backoff

The [client](#client) already does this: on `429` it waits `Retry-After` seconds; on `5xx` it backs off 1, 2, 4, 8, 16 seconds; after five retries it raises. A standalone retry loop is on [Rate limits](https://developers.gocubic.io/rate-limits.md#handling-429).