> ## Documentation Index
> Fetch the complete documentation index at: https://help.vori.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Sync inventory data

> Keep your own system in step with on-hand quantities from the Vori API, with cursor pagination, a first full pass, and incremental polling on updated_at.

This guide walks through building a small client that mirrors on-hand quantities from the Vori API into your own system — the pattern you can use to feed purchasing, replenishment, or an online storefront.

Inventory is different from a transaction feed. A transaction is written once and never changes, so you append it and move on. A product's on-hand quantity changes all day, under an ID that stays the same. What you are building is a mirror you keep up to date, not a log you add to.

## Authenticate with an API key

<Note>
  See the [authentication docs](https://help.vori.com/api/introduction#authentication) for how to create an API key. Make sure the role you assign it can read products.
</Note>

## List inventory

```bash theme={null}
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'
```

The list endpoint supports the following filters, declared as typed query parameters rather than one opaque filter blob:

| Filter                | Type       | Notes                                                                                                                                                 |
| --------------------- | ---------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- |
| `store_id`            | identifier | The store the product belongs to. Repeat the parameter to match up to 20 stores. Leave it off and the endpoint returns every store your key can read. |
| `store_department_id` | identifier | The department the product belongs to. Repeat the parameter to match up to 20 departments.                                                            |
| `updated_at`          | timestamp  | When Vori last wrote the on-hand quantity. Supports `[gt]`/`[gte]`/`[lt]`/`[lte]` range operators                                                     |

Every list response uses one envelope:

```json theme={null}
{
  "data": [ ... ],
  "has_more": true
}
```

Each record is one product's current on-hand quantity — see [the store product inventory object](/api/store-product-inventory/object) for its fields. The `id` is the product ID, so it is what you upsert on in your own table.

## Page through results

Pagination is cursor-based, not offset-based — pass `limit` and `starting_after`/`ending_before` with the `id` of the last record you saw, and keep going while `has_more` is `true`:

```python theme={null}
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
```

Results are always ordered by `id` descending — this order is not client-selectable. A product's `id` never changes, so a quantity that moves while you are paging does not shuffle products between pages.

## Poll for changes

Your first run has nothing to catch up from, so it leaves the `updated_at` filter off and pulls every quantity. After that, ask only for what moved. Track a watermark and filter on `updated_at[gte]` from where the last run left off:

```python theme={null}
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()
```

`updated_at` moves whenever the on-hand quantity changes — a sale, a delivery, a count, an adjustment. Renaming a product or changing its price does not move it, so a poll returns only the products whose quantity actually changed.

The five-minute buffer means each window starts a little before the last one ended, so a quantity written while the previous run was in flight is picked up rather than stepped over. Consecutive runs overlap slightly as a result, which is why your write has to upsert on `id`. Re-reading a product is harmless. Missing one is not.

One run covers every store your API key can read, so keep a single watermark rather than one per store. Every record carries `store_id`, so you can split the results by store in your own system afterwards. Polling store by store multiplies your request count and leaves you several watermarks to keep in step, for no benefit.

## Handle rate limits

Requests are rate-limited per API key. Every response carries your current standing:

* `X-RateLimit-Limit` — max requests allowed in the current window
* `X-RateLimit-Remaining` — requests left in the window

A `429` response includes `Retry-After` (seconds) — back off and retry rather than hammering the endpoint.

Give the retry a bound so a long rate-limit stretch ends in a clear error rather than spinning forever:

```python theme={null}
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")
```

## Putting it together

Everything above in one script. Point `VORI_API_KEY` at your key, replace `write_to_your_inventory_store`, and run it on whatever schedule suits you — cron, Airflow, a worker loop.

The first run finds no watermark and pulls the whole catalog. Every run after that pulls only what changed, so the same script covers both and you do not have to remember which mode you are in.

It writes each page as it arrives rather than collecting everything first, so memory stays flat however large the catalog is, and a run that dies partway keeps the pages it already wrote. The checkpoint records how far through the current window it got and where the next window will start.

<Warning>
  The checkpoint is saved after every page and carries both values, but only the cursor changes as you go — the watermark keeps the previous run's value until the whole window has been written.

  Products come back in `id` order, which says nothing about when each one was counted. A run partway through has covered part of the catalog, not the earliest or latest part of the window. Advancing the watermark there and then failing would leave the next run starting after products it never read, skipping them until something counts them again.
</Warning>

```python theme={null}
"""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()
```

Each row in `data` is the complete current picture for one product, so this script never needs a second call per product. There is no history to walk and no deltas to apply — whatever `current` says is what the shelf is expected to hold right now.

<br />


## Related topics

- [Resolve POS Data Sync Delays](/hardware-and-integrations/cashier-screen-and-shopper-display/resolve-pos-data-sync-delays.md)
- [Re-Sync Missing Product or Invoice Data on a Store Handheld](/product-and-product-catalog-management/product-editing-and-activation/re-sync-missing-product-or-invoice-data-on-a-store-handheld.md)
- [How to export the product catalog](/product-and-product-catalog-management/product-catalog-and-bulk-management/how-to-export-the-product-catalog.md)