Python examples

Basic lookup

import os
import requests

API_KEY = os.environ['RETAILERAPI_KEY']
HEADERS = {'Authorization': f'Bearer {API_KEY}'}

r = requests.get(
    'https://api.retailerapi.com/v1/products/19667262713',
    headers=HEADERS,
    timeout=10,
)
r.raise_for_status()
product = r.json()
print(product['title'], product['current_price'])

Cross-retailer lookup

r = requests.get(
    'https://api.retailerapi.com/v1/products/19667262713',
    params={'include_cross_retailer': 'true'},
    headers=HEADERS,
)
r.raise_for_status()
product = r.json()

# Find cheapest retailer (cross_retailer is now a map keyed by retailer slug)
others = [
    c for slug, c in (product.get('cross_retailer') or {}).items()
    if slug != 'walmart' and c['status'] == 'ok' and isinstance(c.get('price'), (int, float))
]
if others:
    cheapest = min(others, key=lambda c: c['price'])
    print(f"Cheapest: {cheapest['retailer']} at $ {cheapest['price']}")

Reseller view (cross-retailer + seller context)

r = requests.get(
    'https://api.retailerapi.com/v1/products/19667262713',
    params={'include_cross_retailer': 'true', 'include_seller_context': 'true'},
    headers=HEADERS,
)
product = r.json()
wm = product['cross_retailer']['walmart']
sc = wm['seller_context']
print(f"Walmart referral fee: $ {sc['referral_fee_usd']}")
print(f"Walmart WFS fee: $ {sc['wfs_fee_usd']}")
print(f"Restricted on Walmart: {sc['restricted']['flag']}")

Price aggregates (free in base call)

r = requests.get(
    'https://api.retailerapi.com/v1/products/19667262713',
    headers=HEADERS,
)
data = r.json()

# 30d / 60d / 90d / 180d. A window we hold no bars for is None.
w = (data.get('stats') or {}).get('windows', {}).get('90d')
if w:
    print(f"90d low $ {w['low']} - high $ {w['high']} - avg $ {w['average']}")
    # Read days_covered before trusting the label: a 90d low backed by
    # 12 days of data is not the same claim as one backed by 90.
    print(f"based on {w['days_covered']} days of data")

Raw daily bars (+5 tokens)

r = requests.get(
    'https://api.retailerapi.com/v1/products/19667262713',
    params={'include_price_series': 'true'},
    headers=HEADERS,
)
for bar in (r.json().get('price_history') or {}).get('series', []):
    print(bar['date'], bar['open'], bar['high'], bar['low'], bar['close'])

Retry with backoff

import time

def call_with_retry(url, params=None, max_retries=3):
    for attempt in range(max_retries + 1):
        r = requests.get(url, params=params, headers=HEADERS, timeout=10)
        if r.status_code < 500 and r.status_code != 429:
            return r
        if r.status_code == 429:
            retry_after = int(r.headers.get('Retry-After', '5'))
            time.sleep(retry_after)
            continue
        if r.status_code >= 500 and attempt < max_retries:
            time.sleep(2 ** attempt * 0.5)
            continue
        return r

Bulk lookup with thread pool

from concurrent.futures import ThreadPoolExecutor

def lookup_one(upc):
    r = requests.get(
        f'https://api.retailerapi.com/v1/products/{upc}',
        headers=HEADERS,
    )
    return r.json() if r.ok else None

upcs = ['19667262713', '194629116676', '728028502244']
with ThreadPoolExecutor(max_workers=5) as exe:
    results = list(exe.map(lookup_one, upcs))

for upc, product in zip(upcs, results):
    if product:
        title = product["title"][:40]
        price = product["current_price"]
        print(f"{upc}: {title} dollars {price}")

Pydantic models (typed responses)

from pydantic import BaseModel
from typing import Optional, List

class CrossRetailerCell(BaseModel):
    retailer: str
    status: str
    price: Optional[float] = None
    url: Optional[str] = None

class Product(BaseModel):
    item_id: str
    title: str
    brand: Optional[str] = None
    current_price: Optional[float] = None
    offers_count: int = 0
    walmart_url: Optional[str] = None
    cross_retailer: Optional[dict[str, CrossRetailerCell]] = None

product = Product(**r.json())
print(product.title, product.current_price)