Sync your inventory with another system

# Export FILMO_API_KEY first: a key with the scopes listed above.
FILMO_URL="${FILMO_BASE_URL:-https://app.filmo.io}"

// filmo.mjs: a small client. Node 22 or later, no dependencies.
export const API = (process.env.FILMO_BASE_URL ?? "https://app.filmo.io") + "/api/v1";
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

// GET with query parameters. Waits out a 429 as Retry-After says.
export async function get(path, query = {}, headers = {}) {
  const url = new URL(API + path);
  for (const [key, value] of Object.entries(query)) url.searchParams.set(key, String(value));
  for (;;) {
    const res = await fetch(url, {
      headers: { Authorization: ["Bearer", process.env.FILMO_API_KEY].join(" "), ...headers },
    });
    if (res.status === 429) {
      await sleep(Number(res.headers.get("Retry-After") ?? 1) * 1000);
      continue;
    }
    if (res.status === 304) return { status: 304, etag: res.headers.get("ETag") };
    if (!res.ok) {
      throw Object.assign(new Error(res.status + " " + (await res.text())), { status: res.status });
    }
    return {
      ...(await res.json()),
      status: res.status,
      etag: res.headers.get("ETag"),
      serverTime: new Date(res.headers.get("Date")), // the server's clock, not yours
    };
  }
}

# filmo.py: a small client. Python 3 and requests.
import os
import time
from email.utils import parsedate_to_datetime

import requests

API = os.environ.get("FILMO_BASE_URL", "https://app.filmo.io") + "/api/v1"
session = requests.Session()
session.headers["Authorization"] = f"Bearer {os.environ['FILMO_API_KEY']}"


def get(path, headers=None, **query):
    """GET with query parameters. Waits out a 429 as Retry-After says."""
    while True:
        res = session.get(API + path, params=query, headers=headers)
        if res.status_code == 429:
            time.sleep(int(res.headers.get("Retry-After", "1")))
            continue
        if res.status_code == 304:
            return {"status": 304, "etag": res.headers.get("ETag")}
        res.raise_for_status()
        body = res.json()
        body["status"] = res.status_code
        body["etag"] = res.headers.get("ETag")
        body["serverTime"] = parsedate_to_datetime(res.headers["Date"])  # the server's clock
        return body

# Export FILMO_API_KEY first: a key with the scopes listed above.
FILMO_URL="${FILMO_BASE_URL:-https://app.filmo.io}"

curl -sS "$FILMO_URL/api/v1/account" \
  -H "Authorization: Bearer $FILMO_API_KEY"

import { get } from "./filmo.mjs";

const { data: account } = await get("/account");
console.log(account.timezone, account.currency, account.rateLimits.global.perMinute);

from filmo import get

account = get("/account")["data"]
print(account["timezone"], account["currency"], account["rateLimits"]["global"]["perMinute"])

FILMO_URL="${FILMO_BASE_URL:-https://app.filmo.io}"

# One window [updatedFrom, updatedBefore), one page. If pagination.hasMore is true, split the window.
curl -sS -G "$FILMO_URL/api/v1/items" \
  -H "Authorization: Bearer $FILMO_API_KEY" \
  --data-urlencode "updatedFrom=2026-09-01T00:00:00.000Z" \
  --data-urlencode "updatedBefore=2026-10-01T00:00:00.000Z" \
  --data-urlencode "sort=updatedAt" \
  --data-urlencode "limit=500"

// window.mjs: the windowed read of the syncing guide, one page per query.
import { get } from "./filmo.mjs";

const iso = (ms) => new Date(ms).toISOString();

export const START = Date.parse("2020-01-01T00:00:00Z"); // move it to the month of your first record

// B: the server's time minus the safety margin of the guide (2 minutes and 1 second). The time is the
// Date header of a response, never your own clock.
export async function upperBound() {
  const { serverTime } = await get("/account");
  return serverTime.getTime() - (2 * 60 * 1000 + 1000);
}

// Every record whose updatedAt (or createdAt) falls in [from, to), in epoch milliseconds.
// It never follows the cursor: a window that overflows one page is split in half.
export async function windowed(path, from, to, { field = "updated", query = {} } = {}) {
  const page = await get(path, {
    ...query,
    [field + "From"]: iso(from),
    [field + "Before"]: iso(to),
    sort: field + "At",
    limit: 500,
  });
  if (!page.pagination.hasMore) return page.data;
  if (to - from <= 1) throw new Error("More than 500 records share one millisecond: see the guide");
  const mid = Math.floor((from + to) / 2);
  const merged = new Map();
  for (const half of [
    await windowed(path, from, mid, { field, query }),
    await windowed(path, mid, to, { field, query }),
  ]) {
    for (const record of half) merged.set(record.id, record); // deduplicate by id
  }
  return [...merged.values()];
}

// Walks [start, end) in 30-day windows. Returns a Map of id to record.
export async function readRange(path, start, end, options) {
  const step = 30 * 24 * 3600 * 1000;
  const records = new Map();
  for (let from = start; from < end; from += step) {
    for (const record of await windowed(path, from, Math.min(from + step, end), options)) {
      records.set(record.id, record);
    }
  }
  return records;
}

# window.py: the windowed read of the syncing guide, one page per query.
from datetime import datetime, timezone

from filmo import get


START = int(datetime(2020, 1, 1, tzinfo=timezone.utc).timestamp() * 1000)  # move it to the month of your first record


def upper_bound():
    """B: the server's time minus the safety margin of the guide (2 minutes and 1 second).
    The time is the Date header of a response, never your own clock."""
    return int(get("/account")["serverTime"].timestamp() * 1000) - (2 * 60 * 1000 + 1000)


def iso(ms):
    stamp = datetime.fromtimestamp(ms / 1000, tz=timezone.utc)
    return stamp.isoformat(timespec="milliseconds").replace("+00:00", "Z")


def windowed(path, start, end, field="updated", **query):
    """Every record whose updatedAt (or createdAt) falls in [start, end), in epoch ms.
    It never follows the cursor: a window that overflows one page is split in half."""
    page = get(
        path,
        **query,
        **{field + "From": iso(start), field + "Before": iso(end), "sort": field + "At", "limit": 500},
    )
    if not page["pagination"]["hasMore"]:
        return page["data"]
    if end - start <= 1:
        raise RuntimeError("More than 500 records share one millisecond: see the guide")
    mid = (start + end) // 2
    merged = {r["id"]: r for r in windowed(path, start, mid, field, **query)}
    merged.update({r["id"]: r for r in windowed(path, mid, end, field, **query)})
    return list(merged.values())


def read_range(path, start, end, field="updated", **query):
    """Walks [start, end) in 30-day windows. Returns a dict of id to record."""
    step = 30 * 24 * 3600 * 1000
    records = {}
    for begin in range(start, end, step):
        for record in windowed(path, begin, min(begin + step, end), field, **query):
            records[record["id"]] = record
    return records

FILMO_URL="${FILMO_BASE_URL:-https://app.filmo.io}"

# One window [updatedFrom, updatedBefore), one page. If pagination.hasMore is true, split the window.
curl -sS -G "$FILMO_URL/api/v1/items" \
  -H "Authorization: Bearer $FILMO_API_KEY" \
  --data-urlencode "updatedFrom=2026-09-01T00:00:00.000Z" \
  --data-urlencode "updatedBefore=2026-10-01T00:00:00.000Z" \
  --data-urlencode "sort=updatedAt" \
  --data-urlencode "limit=500"

import { readRange, START, upperBound } from "./window.mjs";

const B = await upperBound(); // the end of this load; save it as the watermark when every list is read
const items = await readRange("/items", START, B);
console.log(items.size + " items");

from window import START, read_range, upper_bound

B = upper_bound()  # the end of this load; save it as the watermark when every list is read
items = read_range("/items", START, B)
print(len(items), "items")

# The units of one catalog item, with where each one is committed right now.
curl -sS -G "$FILMO_URL/api/v1/serialized-items" \
  -H "Authorization: Bearer $FILMO_API_KEY" \
  --data-urlencode "itemId=$ITEM_ID" \
  --data-urlencode "limit=500"

import { readRange, START, upperBound } from "./window.mjs";

const B = await upperBound();
const units = await readRange("/serialized-items", START, B);
for (const unit of units.values()) {
  // unit.currentOrderId is the order that has prepared, picked up, sold or got the unit back.
  // It is null otherwise, also when the unit is only reserved on an order.
  console.log(unit.code, unit.status, unit.itemId, unit.currentOrderId);
}

from window import START, read_range, upper_bound

B = upper_bound()
units = read_range("/serialized-items", START, B)
for unit in units.values():
    # unit["currentOrderId"] is the order that has prepared, picked up, sold or got the unit back.
    # It is None otherwise, also when the unit is only reserved on an order.
    print(unit["code"], unit["status"], unit["itemId"], unit["currentOrderId"])

# The lines of one order, including the ones added during delivery.
curl -sS -G "$FILMO_URL/api/v1/orders/$ORDER_ID/lines" \
  -H "Authorization: Bearer $FILMO_API_KEY" \
  --data-urlencode "includeDelivery=true" \
  --data-urlencode "limit=500"

// orders.mjs
import { get } from "./filmo.mjs";

// Lines are only reachable through their order, and they have no updatedAt.
export async function readLines(orderId) {
  const lines = [];
  let cursor;
  do {
    const page = await get("/orders/" + orderId + "/lines", {
      includeDelivery: true,
      limit: 500,
      ...(cursor && { cursor }),
    });
    lines.push(...page.data);
    cursor = page.pagination.hasMore ? page.pagination.nextCursor : undefined;
  } while (cursor);
  return lines;
}

# orders.py
from filmo import get


def read_lines(order_id):
    """Lines are only reachable through their order, and they have no updatedAt."""
    lines, cursor = [], None
    while True:
        query = {"includeDelivery": "true", "limit": 500}
        if cursor:
            query["cursor"] = cursor
        page = get("/orders/" + order_id + "/lines", **query)
        lines += page["data"]
        if not page["pagination"]["hasMore"]:
            return lines
        cursor = page["pagination"]["nextCursor"]

import { readRange, START, upperBound } from "./window.mjs";
import { readLines } from "./orders.mjs";

const B = await upperBound();
const orders = await readRange("/orders", START, B);
for (const order of orders.values()) {
  try {
    const lines = await readLines(order.id);
    // Replace the lines you hold for order.id with these.
    for (const line of lines) console.log(order.orderNumber, line.itemId, line.quantity, line.serials.length);
  } catch (error) {
    if (error.status !== 404) throw error; // 404: archived or deleted since the list was read
  }
}

import requests

from orders import read_lines
from window import START, read_range, upper_bound

B = upper_bound()
orders = read_range("/orders", START, B)
for order in orders.values():
    try:
        for line in read_lines(order["id"]):
            # Replace the lines you hold for this order with these.
            print(order["orderNumber"], line["itemId"], line["quantity"], len(line["serials"]))
    except requests.HTTPError as error:
        if error.response.status_code != 404:  # 404: archived or deleted since the list was read
            raise

// poll.mjs: one incremental pass. Run it every few minutes.
import { get } from "./filmo.mjs";
import { readRange } from "./window.mjs";
import { readLines } from "./orders.mjs";

const MARGIN = 2 * 60 * 1000 + 1000; // the guide's safety margin: 2 minutes and 1 second

export async function pollOnce(state) {
  // state = { watermark } in epoch ms, kept in your database or a file.
  const { serverTime } = await get("/account");
  const A = state.watermark;
  const B = serverTime.getTime() - MARGIN;
  if (B <= A) return state;

  const items = await readRange("/items", A, B);
  const units = await readRange("/serialized-items", A, B);
  const orders = await readRange("/orders", A, B);

  // ...upsert items and units by id here...

  // Lines have no updatedAt: re-read the lines of every order that appears in the window.
  for (const order of orders.values()) {
    const lines = await readLines(order.id);
    // ...replace the lines you hold for order.id...
  }
  return { watermark: B }; // moves only when every list above was read
}

# poll.py: one incremental pass. Run it every few minutes.
from filmo import get
from window import read_range
from orders import read_lines

MARGIN = 2 * 60 * 1000 + 1000  # the guide's safety margin: 2 minutes and 1 second


def poll_once(state):
    """state = {"watermark": epoch ms}, kept in your database or a file."""
    server_time = get("/account")["serverTime"]
    a = state["watermark"]
    b = int(server_time.timestamp() * 1000) - MARGIN
    if b <= a:
        return state

    items = read_range("/items", a, b)
    units = read_range("/serialized-items", a, b)
    orders = read_range("/orders", a, b)

    # ...upsert items and units by id here...

    # Lines have no updatedAt: re-read the lines of every order that appears in the window.
    for order in orders.values():
        lines = read_lines(order["id"])
        # ...replace the lines you hold for this order...
    return {"watermark": b}  # moves only when every list above was read

# A week later: list a createdAt window...
curl -sS -G "$FILMO_URL/api/v1/items" \
  -H "Authorization: Bearer $FILMO_API_KEY" \
  --data-urlencode "createdFrom=2026-09-01T00:00:00.000Z" \
  --data-urlencode "createdBefore=2026-10-01T00:00:00.000Z" \
  --data-urlencode "sort=createdAt" \
  --data-urlencode "limit=500"

# ...and confirm every id you hold that is missing from it. Only a 404 counts.
curl -sS -o /dev/null -w "%{http_code}\n" "$FILMO_URL/api/v1/items/$MISSING_ITEM_ID" \
  -H "Authorization: Bearer $FILMO_API_KEY"

// sweep.mjs: weekly reconciliation of one list.
import { get } from "./filmo.mjs";
import { readRange } from "./window.mjs";

// held: Map of id to the updatedAt you stored. Returns what to re-read and what is gone.
export async function sweep(path, held, start, end) {
  const live = await readRange(path, start, end, { field: "created" });

  // Updates the early passes may have skipped: the stored updatedAt differs.
  const stale = [...live].filter(([id, record]) => held.get(id) !== record.updatedAt).map(([id]) => id);

  // A missing id is not a deletion until its own GET answers 404.
  const gone = [];
  for (const id of held.keys()) {
    if (live.has(id)) continue;
    try {
      await get(path + "/" + id); // still there: the sweep skipped it, do not delete
    } catch (error) {
      if (error.status !== 404) throw error;
      gone.push(id); // for an order, 404 means archived or deleted
    }
  }
  return { stale, gone };
}

# sweep.py: weekly reconciliation of one list.
import requests

from filmo import get
from window import read_range


def sweep(path, held, start, end):
    """held: dict of id to the updatedAt you stored. Returns what to re-read and what is gone."""
    live = read_range(path, start, end, field="created")

    # Updates the early passes may have skipped: the stored updatedAt differs.
    stale = [i for i, record in live.items() if held.get(i) != record["updatedAt"]]

    # A missing id is not a deletion until its own GET answers 404.
    gone = []
    for record_id in held:
        if record_id in live:
            continue
        try:
            get(path + "/" + record_id)  # still there: the sweep skipped it, do not delete
        except requests.HTTPError as error:
            if error.response.status_code != 404:
                raise
            gone.append(record_id)  # for an order, 404 means archived or deleted
    return {"stale": stale, "gone": gone}