priceq API

curl "https://app.priceq.io/api/v1/skus?limit=3" \
  -H "Authorization: Bearer $PRICEQ_KEY" \
  -H "Accept-Language: en"

import os, requests

r = requests.get(
    "https://app.priceq.io/api/v1/skus",
    params={"limit": 3},
    headers={"Authorization": f"Bearer {os.environ['PRICEQ_KEY']}"},
    timeout=30,
)
r.raise_for_status()
print(r.json()["total"], "products")

const res = await fetch(
  "https://app.priceq.io/api/v1/skus?limit=3",
  { headers: { Authorization: `Bearer ${process.env.PRICEQ_KEY}` } }
);
if (!res.ok) throw new Error(`priceq ${res.status}`);
const { items, total } = await res.json();
console.log(total, "products");

# PowerShell has no real curl — use Invoke-RestMethod
$h = @{ Authorization = "Bearer $env:PRICEQ_KEY" }
$r = Invoke-RestMethod "https://app.priceq.io/api/v1/skus?limit=3" -Headers $h
$r.items | Format-Table sku, name, own_price_cents

{
  "items": [
    {
      "id": "67469b7c-0cd6-48b9-8753-89ddf9beb978",
      "sku": "FJ-FLASK-500",
      "ean": "8712345678906",
      "name": "Fjordlund vacuum flask 500 ml",
      "brand": "Fjordlund",
      "category": "drinkware",
      "status": "active",
      "source": "api",
      "sync_status": "active",
      "own_price_cents": 2495,
      "updated_at": "2026-09-10T06:33:13Z"
    }
  ],
  "total": 5
}

Authorization: Bearer pq_a1b2c3d4_your-secret-part

POST /api-keys
{ "name": "ERP nightly", "scope": "ingest", "expires_in_days": 90 }

{
  "detail": "Not signed in. Send your key as `Authorization: Bearer pq_...`."
}

# allowed
GET    /skus  /mappings  /alerts  /trends/*
POST   /catalog/products  /catalog/sync
POST   /alerts/{id}/read  /alerts/scan
GET    /exports/report/{type}
*      /webhooks

# 403 — do this in the app instead
GET    /exports/full
GET    /audit-log
*      /api-keys  /team  /me/subscription

def all_products(session, key):
    offset, out = 0, []
    while True:
        r = session.get(
            "https://app.priceq.io/api/v1/skus",
            params={"limit": 1000, "offset": offset},
            headers={"Authorization": f"Bearer {key}"},
            timeout=60,
        )
        r.raise_for_status()
        page = r.json()
        out += page["items"]
        offset += 1000
        if offset >= page["total"]:
            return out

async function allProducts(key) {
  const out = [];
  let offset = 0, total = Infinity;
  while (offset < total) {
    const res = await fetch(
      `https://app.priceq.io/api/v1/skus?limit=1000&offset=${offset}`,
      { headers: { Authorization: `Bearer ${key}` } }
    );
    if (!res.ok) throw new Error(`priceq ${res.status}`);
    const page = await res.json();
    out.push(...page.items);
    total = page.total;
    offset += 1000;
  }
  return out;
}

curl -G "https://app.priceq.io/api/v1/skus" \
  --data-urlencode "category=gpu,psu" \
  --data-urlencode "brand=Fjordlund" \
  --data-urlencode "limit=100" \
  -H "Authorization: Bearer $PRICEQ_KEY"

[
  {
    "country": "NL",
    "currency": "EUR",
    "msrp_cents": 59999,
    "sell_price_cents": 2495,
    "cost_cents": 41000,
    "min_price_cents": 44900,
    "max_price_cents": null,
    "updated_at": "2026-09-10T06:33:13Z"
  }
]

{
  "items": [
    {
      "id": "3f0c…",
      "sku_id": "67469b7c…",
      "listing_url": "https://azerty.nl/p/…",
      "state": "matched",
      "confidence": 97,
      "bucket": "green",
      "flags": [],
      "last_observed_at": "2026-09-10T04:12:08Z",
      "retailer": { "slug": "azerty" }
    }
  ],
  "total": 1
}

curl "https://app.priceq.io/api/v1/alerts?status=unread&sort=impact" \
  -H "Authorization: Bearer $PRICEQ_KEY"

{
  "items": [
    {
      "id": "9c2b…",
      "alert_type": "competitor_cheaper",
      "severity": "warning",
      "status": "unread",
      "title": "Azerty is 4% goedkoper",
      "sku_id": "67469b7c…",
      "impact_score": 82.4,
      "created_at": "2026-09-10T05:02:00Z"
    }
  ],
  "total": 1,
  "unread_count": 1
}

curl -L "https://app.priceq.io/api/v1/exports/report/prijspositie" \
  -G --data-urlencode "from=2026-08-01T00:00:00Z" \
     --data-urlencode "to=2026-09-01T00:00:00Z" \
  -H "Authorization: Bearer $PRICEQ_KEY" \
  -H "Accept-Language: en" \
  -o price-position.xlsx

{
  "plan": "start",
  "limits":   { "skus": 200, "retailers": 3, "users": 2 },
  "usage":    { "skus": 64,  "retailers": 0, "users": 0 },
  "features": { "price_proposals": false },
  "scans_per_day": 1,
  "trial_ends_at": null
}

curl -X POST "https://app.priceq.io/api/v1/catalog/products" \
  -H "Authorization: Bearer $PRICEQ_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "currency": "EUR",
    "products": [
      {
        "id": "ERP-1",
        "name": "Fjordlund vacuum flask 500 ml",
        "ean": "8712345678906",
        "sku": "FJ-FLASK-500",
        "brand": "Fjordlund",
        "category": "Drinkware",
        "price": 2495,
        "list_price": 59999,
        "cost_price": 41000
      }
    ]
  }'

import os, requests

payload = {
    "currency": "EUR",
    "products": [{
        "id": "ERP-1",
        "name": "Fjordlund vacuum flask 500 ml",
        "ean": "8712345678906",
        "price": 2495,
    }],
}
r = requests.post(
    "https://app.priceq.io/api/v1/catalog/products",
    json=payload,
    headers={"Authorization": f"Bearer {os.environ['PRICEQ_KEY']}"},
    timeout=120,
)
body = r.json()
for item in body["results"]:
    if item["status"] == "failed":
        print(item["id"], item["detail"])
    for w in item["warnings"]:
        print("warn", item["id"], w)

const res = await fetch(
  "https://app.priceq.io/api/v1/catalog/products",
  {
    method: "POST",
    headers: {
      Authorization: `Bearer ${process.env.PRICEQ_KEY}`,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({
      currency: "EUR",
      products: [{ id: "ERP-1", name: "Vacuum flask 500 ml", price: 2495 }],
    }),
  }
);
const body = await res.json();
body.results
  .filter((r) => r.status === "failed")
  .forEach((r) => console.error(r.id, r.detail));

// first delivery
{ "id": "ERP-1", "name": "Vacuum flask 500 ml",
  "ean": "8712345678906", "brand": "Fjordlund" }

// price-only update — ean and brand survive
{ "id": "ERP-1", "name": "Vacuum flask 500 ml", "price": 2295 }
// → ean 8712345678906, brand Fjordlund

// explicit clear
{ "id": "ERP-1", "name": "Vacuum flask 500 ml", "brand": "" }
// → brand removed

"price": 1999      // ✓ € 19,99
"price": "1999"    // ✓ digits in a string
"price": null      // ✓ leaves it unchanged

"price": 12.34     // ✗ no decimals
"price": "12,34"   // ✗ no comma
"price": -5        // ✗ not negative

{
  "summary": {
    "received": 3, "created": 1, "updated": 1,
    "failed": 1, "skipped": 0, "warned": 1
  },
  "results": [
    {
      "index": 0,
      "id": "ERP-1",
      "status": "created",
      "detail": null,
      "warnings": [
        "EAN 8712345678906 is already on 'ERP-7' in your catalogue…"
      ]
    },
    {
      "index": 1, "id": "ERP-2",
      "status": "updated",
      "detail": null, "warnings": []
    },
    {
      "index": 2, "id": "ERP-3",
      "status": "failed",
      "detail": "'price' must be a whole number of cents…",
      "warnings": []
    }
  ],
  "sync_id": null
}

# results[i].index matches products[i] — always
sent = payload["products"]
for r in body["results"]:
    original = sent[r["index"]]
    if r["status"] == "failed":
        log.error("rejected %s: %s", original["id"], r["detail"])

{
  "index": 0,
  "id": "ERP-9",
  "status": "updated",
  "detail": null,
  "warnings": [
    "EAN '8712345678900' is not a valid EAN-13, UPC-12 or"
    " GTIN-14 (wrong check digit or length) and was not stored."
    " The EAN already on the product stays."
  ]
}

"warnings": [
  "EAN 8712345678906 is also on 'E-2' in this request."
  " We match retailer pages on EAN, so these products will"
  " receive each other's prices."
]

"warnings": [
  "Article code 'ART-1' is also on 'T-1' in this request."
  " An article code has to be unique, so this product"
  " becomes 'ART-1-T-2'."
]

import os, requests

API = "https://app.priceq.io/api/v1"
H = {"Authorization": f"Bearer {os.environ['PRICEQ_KEY']}"}
s = requests.Session()

# 1 — open
sync = s.post(f"{API}/catalog/sync", headers=H, timeout=30).json()
sid = sync["sync_id"]

# 2 — deliver in pages of 500
for i in range(0, len(catalogue), 500):
    page = catalogue[i:i + 500]
    r = s.post(
        f"{API}/catalog/sync/{sid}/products",
        json={"currency": "EUR", "products": page},
        headers={**H, "Idempotency-Key": f"{sid}-{i}"},
        timeout=120,
    )
    for item in r.json()["results"]:
        if item["status"] == "failed":
            print("rejected", item["id"], item["detail"])

# 3 — complete; this is the step that retires
done = s.post(f"{API}/catalog/sync/{sid}/complete", headers=H, timeout=60).json()
if done["state"] == "aborted":
    raise SystemExit(f"sync refused: {done['detail']}")
print(f"retired {done['orphaned']} products")

# 1 — open
SID=$(curl -s -X POST "$API/catalog/sync" \
  -H "Authorization: Bearer $PRICEQ_KEY" \
  | sed -n 's/.*"sync_id":"\([^"]*\)".*/\1/p')

# 2 — one page (repeat for each chunk of 500)
curl -s -X POST "$API/catalog/sync/$SID/products" \
  -H "Authorization: Bearer $PRICEQ_KEY" \
  -H "Content-Type: application/json" \
  --data @page-001.json

# 3 — complete
curl -s -X POST "$API/catalog/sync/$SID/complete" \
  -H "Authorization: Bearer $PRICEQ_KEY"

{
  "sync_id": "d76839a6-bdf9-40bd-b133-c0903eb18f4e",
  "state": "completed",
  "started_at": "2026-09-10T02:00:00Z",
  "received": 19842,
  "created": 12,
  "updated": 19830,
  "failed": 0,
  "skipped": 0,
  "orphaned": 158,
  "detail": null,
  "completed_at": "2026-09-10T02:01:07Z"
}

{
  "state": "aborted",
  "orphaned": 0,
  "detail": "This sync would mark 4021 of your 5000 products as"
            " gone. That is too many to do unchecked, so nothing"
            " was changed."
}

curl -X POST "$API/catalog/products" \
  -H "Authorization: Bearer $PRICEQ_KEY" \
  -H "Idempotency-Key: nightly-2026-09-10-page-3" \
  -H "Content-Type: application/json" \
  --data @page-003.json

{
  "detail": "This Idempotency-Key was already used for a request"
            " with different content. Use a new key."
}

import os, itertools, requests

API = "https://app.priceq.io/api/v1"
H = {"Authorization": f"Bearer {os.environ['PRICEQ_KEY']}"}

changed = [
    {"id": p.code, "name": p.name, "price": p.cents}
    for p in erp.products_changed_since(last_run)
]

def chunks(seq, n=500):
    it = iter(seq)
    while (batch := list(itertools.islice(it, n))):
        yield batch

for i, batch in enumerate(chunks(changed)):
    r = requests.post(
        f"{API}/catalog/products",
        json={"currency": "EUR", "products": batch},
        headers={**H, "Idempotency-Key": f"delta-{run_id}-{i}"},
        timeout=120,
    )
    if r.status_code == 429:
        time.sleep(int(r.headers["Retry-After"]))
        continue
    for item in r.json()["results"]:
        if item["warnings"]:
            log.warning("%s: %s", item["id"], item["warnings"])

{
  "detail": [
    {
      "type": "too_long",
      "loc": ["body", "products"],
      "msg": "List should have at most 500 items"
    }
  ]
}

{
  "detail": {
    "code": "plan_limit",
    "resource": "skus",
    "limit": 200,
    "current": 200,
    "plan": "start"
  }
}

HTTP/1.1 429 Too Many Requests
x-ratelimit-limit: 60
x-ratelimit-remaining: 0
x-ratelimit-reset: 1789048083
retry-after: 45

{
  "detail": "Too many requests (60 per 1 minute). The Retry-After"
            " header says how many seconds to wait."
}

def post_with_backoff(session, url, **kw):
    for attempt in range(5):
        r = session.post(url, **kw)
        if r.status_code != 429:
            return r
        time.sleep(int(r.headers.get("Retry-After", 60)))
    r.raise_for_status()

curl -X POST "$API/webhooks" \
  -H "Authorization: Bearer $PRICEQ_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://erp.example.com/hooks/priceq",
    "description": "Nightly sync result",
    "event_types": ["catalog.sync.completed"]
  }'

import base64, hmac, hashlib, time

def verify(secret, headers, body, tolerance=300):
    ts = headers["webhook-timestamp"]
    if abs(time.time() - int(ts)) > tolerance:
        return False   # too old: a replay
    signed = f"{headers['webhook-id']}.{ts}.".encode() + body
    mine = base64.b64encode(
        hmac.new(secret.encode(), signed, hashlib.sha256).digest()
    ).decode()
    theirs = [s.removeprefix("v1,")
              for s in headers["webhook-signature"].split(" ")]
    return any(hmac.compare_digest(mine, t) for t in theirs)

webhook-id: 3f2a…
webhook-timestamp: 1789048083
webhook-signature: v1,K2h…   (two during a rotation)
X-priceq-Signature: sha256=9c4e…
X-priceq-Event: catalog.sync.completed
X-priceq-Delivery-Id: 3f2a…

catalog.api_delivered   ·  api_key  ·  06:31:56
  received 500 · created 12 · updated 488
  failed 0 · skipped 0 · warned 3

catalog.api_sync_completed  ·  api_key  ·  02:01:07
  received 19842 · orphaned 158

# good — ignores fields it does not know
price = item["own_price_cents"]

# fragile — breaks when we add a field
id_, sku, ean, name = item.values()