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 non-Walmart retailer
others = [
    c for c in (product.get('cross_retailer') or [])
    if 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']}")

Price history

r = requests.get(
    'https://api.retailerapi.com/v1/products/19667262713/history',
    params={'timeframe': '90d', 'retailer': 'walmart'},
    headers=HEADERS,
)
data = r.json()
prices = [obs['price'] for obs in data['observations']]
print(f"90d range: $ {min(prices)} - $ {max(prices)}")

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[List[CrossRetailerCell]] = None

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