Sync inventory data

curl -G "https://api.vori.com/v1/store-product-inventory" \
  -H "Authorization: Bearer sk_live_<secret>" \
  --data-urlencode 'updated_at[gte]=2026-08-01T00:00:00Z' \
  --data-urlencode 'limit=100'

{
  "data": [ ... ],
  "has_more": true
}

import requests

BASE = "https://api.vori.com/v1/store-product-inventory"
HEADERS = {"Authorization": "Bearer sk_live_<secret>"}

def fetch_all_inventory(updated_at_gte=None):
    params = {"limit": 100}
    if updated_at_gte:
        params["updated_at[gte]"] = updated_at_gte

    records = []

    while True:
        resp = requests.get(BASE, headers=HEADERS, params=params)
        resp.raise_for_status()
        body = resp.json()
        records.extend(body["data"])

        if not body["has_more"]:
            break
        params["starting_after"] = body["data"][-1]["id"]

    return records

from datetime import datetime, timedelta, timezone

def run_periodic_sync():
    # Load from wherever you persist it — a database row, a config
    # store, a file.
    last_watermark = load_watermark()

    # Where the next run starts. Taken before fetching, not from the max
    # updated_at seen — a quantity written mid-run can land behind that value
    # and would be missed on the next poll.
    next_watermark = (
        datetime.now(timezone.utc) - timedelta(minutes=5)
    ).isoformat()

    records = fetch_all_inventory(last_watermark)
    write_to_your_inventory_store(records)

    # Persist only after the write succeeds. If write_to_your_inventory_store
    # raises, this line is never reached, so the stored watermark stays put
    # and the next run retries the same window instead of skipping it.
    save_watermark(next_watermark)

    print(f"Synced {len(records)} products since {last_watermark}")

if __name__ == "__main__":
    run_periodic_sync()

import time

MAX_RETRIES = 5

def get_with_backoff(url, **kwargs):
    for _ in range(MAX_RETRIES):
        resp = requests.get(url, **kwargs)
        if resp.status_code != 429:
            resp.raise_for_status()
            return resp
        # Wait exactly as long as the response asks rather than guessing.
        time.sleep(int(resp.headers.get("Retry-After", "5")))

    raise RuntimeError(f"still rate limited after {MAX_RETRIES} attempts")

"""Mirror Vori inventory into your own system.

The first run pulls every product. Each run after that pulls only the products
whose on-hand quantity changed since the last successful run, writing each page
as it arrives so nothing accumulates in memory and an interrupted run resumes
where it stopped.
"""

import json
import os
import time
from datetime import datetime, timedelta, timezone
from pathlib import Path

import requests

BASE = "https://api.vori.com/v1/store-product-inventory"
HEADERS = {"Authorization": f"Bearer {os.environ['VORI_API_KEY']}"}
PAGE_SIZE = 100
MAX_RETRIES = 5

# A record's timestamp is set when Vori writes it and it becomes readable a
# moment later, so each run starts a little before the last one did rather than
# exactly where it ended.
WATERMARK_BUFFER = timedelta(minutes=5)

# One checkpoint for the whole key. A run covers every store the key can read,
# and each record carries store_id if you want to split them later.
CHECKPOINT_FILE = Path(
    os.environ.get("VORI_CHECKPOINT_FILE", "vori-inventory-checkpoint.json")
)


def load_checkpoint():
    if CHECKPOINT_FILE.exists():
        return json.loads(CHECKPOINT_FILE.read_text())
    # No watermark on the first run, which is what makes it a full pass.
    return {}


def save_checkpoint(checkpoint):
    # Write then rename, so a crash mid-write cannot leave a half-written file
    # that the next run fails to parse.
    tmp = CHECKPOINT_FILE.with_suffix(".tmp")
    tmp.write_text(json.dumps(checkpoint))
    tmp.replace(CHECKPOINT_FILE)


def get_with_backoff(params):
    for _ in range(MAX_RETRIES):
        resp = requests.get(BASE, headers=HEADERS, params=params, timeout=30)
        if resp.status_code != 429:
            resp.raise_for_status()
            return resp
        # Wait exactly as long as the response asks rather than guessing.
        time.sleep(int(resp.headers.get("Retry-After", "5")))

    raise RuntimeError(f"still rate limited after {MAX_RETRIES} attempts")


def iter_pages(updated_at_gte=None, starting_after=None):
    """Yield each page of inventory written at or after `updated_at_gte`.

    Pass no lower bound to walk every product. The bound is inclusive, so a
    watermark can be re-used verbatim as the next run's lower bound without
    losing anything on the boundary.
    """
    params = {"limit": PAGE_SIZE}
    if updated_at_gte:
        params["updated_at[gte]"] = updated_at_gte
    if starting_after:
        params["starting_after"] = starting_after

    while True:
        body = get_with_backoff(params).json()
        page = body["data"]

        if page:
            yield page

        if not body["has_more"]:
            return

        # A cursor, not an offset: the id of the last row on the page just read.
        params["starting_after"] = page[-1]["id"]


def write_to_your_inventory_store(records):
    """Replace this with your own write.

    Key on `id` and upsert. A product's row is a current value, not an event, so
    the newest read wins and re-reading one costs you nothing.
    """
    for record in records:
        print(record)


def run_sync():
    checkpoint = load_checkpoint()

    # Absent on the first run, so that run asks for no lower bound and pulls
    # the whole catalog.
    watermark = checkpoint.get("watermark")

    # Where the next run will start: this run's start time, less the buffer.
    # Taken before fetching, not after — a quantity written mid-run can land
    # behind it and would be missed next time. A resumed run keeps the value the
    # interrupted one chose, for the same reason.
    next_watermark = checkpoint.get("next_watermark") or (
        (datetime.now(timezone.utc) - WATERMARK_BUFFER)
        .isoformat()
        .replace("+00:00", "Z")
    )

    count = 0

    for page in iter_pages(watermark, checkpoint.get("cursor")):
        write_to_your_inventory_store(page)
        count += len(page)

        # Saved only after the page is written, and recording the cursor rather
        # than the clock: pages arrive in id order, so a run that has read three
        # pages holds part of the catalog and not the rest. Moving the watermark
        # here would skip every product still to come.
        save_checkpoint(
            {
                "watermark": watermark,
                "next_watermark": next_watermark,
                "cursor": page[-1]["id"],
            }
        )

    # The whole window has been written, so the watermark can move and the
    # cursor is no longer needed.
    save_checkpoint({"watermark": next_watermark})

    print(f"Synced {count} products since {watermark or 'the beginning'}")


if __name__ == "__main__":
    run_sync()