Bulk ASIN Lookup: Fetch Thousands of Amazon Products Efficiently

A bulk ASIN lookup through the SellerMagnet API is one request per unique ASIN, so the whole job comes down to paying for each product exactly once. Normalise and deduplicate the input, skip ASINs you have already fetched, run a bounded pool of workers, retry only transient errors, and record products that do not exist instead of asking again. The Python script below does all of that and survives a restart.

What makes a bulk ASIN lookup expensive?

Real-world ASIN lists are messy. Spreadsheet exports repeat the same product across rows, mix upper and lower case, carry stray spaces, and include values that are not ASINs at all. Every duplicate that reaches the API is a credit spent on data you already have, and every crash that restarts the job from the top pays for the finished part twice.

The fix is a pipeline where each stage removes work before it reaches the network. The table traces an example input of 10,000 rows; your ratios will differ, but the shape is typical.

How do I clean and deduplicate an ASIN list?

An ASIN is Amazon's 10-character product identifier made of capital letters and digits. Upper-case and trim every value, keep only those that match that shape, and deduplicate while preserving the original order. Values that fail the pattern would be rejected with HTTP 400 anyway; dropping them locally just saves the round trip. If your list holds EANs, UPCs or ISBNs instead, convert them first with the ASIN converter.

clean.py - normalise, validate and deduplicate in one pass

import csv
import re

ASIN_RE = re.compile(r"^[A-Z0-9]{10}$")


def load_asins(path: str, column: str = "asin") -> list[str]:
    seen, asins = set(), []
    with open(path, newline="", encoding="utf-8-sig") as handle:
        for row in csv.DictReader(handle):
            asin = (row.get(column) or "").strip().upper()
            if ASIN_RE.match(asin) and asin not in seen:
                seen.add(asin)
                asins.append(asin)
    return asins

How many ASINs can I look up in parallel?

Each API key has a cap on concurrent requests. A short burst above it is queued for up to 20 seconds; only sustained overload returns HTTP 429. A pool of 5 to 10 threads keeps throughput high without touching the cap, and the retry rules from the error-handling guide absorb the occasional 429 or 503. Throughput also has a ceiling per client IP - 600 requests a minute and 15,000 an hour as of September 2026 - so a job of more than 15,000 ASINs takes over an hour from one machine.

How do I make a bulk lookup resumable?

Write each result to a JSON Lines file - one JSON object per line - the moment it arrives. On start, read the file and skip every ASIN already in it, including those recorded as missing. A crash, a Ctrl-C or a laptop lid then costs at most the handful of requests in flight: the queue is dropped, and the next run continues where the last one stopped.

bulk_lookup.py - bounded pool, capped paid retries, resume

import json
import os
import random
import sys
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

import requests

from clean import load_asins  # clean.py, above

URL = "https://sellermagnet-api.com/api/amazon-product-lookup"
OUT = "results.jsonl"
WORKERS = 8
MAX_PAID_RETRIES = 2  # 502/503 are billed per attempt; a 429 is free


class OutOfCredits(Exception):
    pass


def fetch(asin: str, marketplace: str) -> dict:
    params = {"asin": asin, "marketplaceId": marketplace,
              "api_key": os.environ["SELLERMAGNET_API_KEY"]}
    paid_retries = 0
    for attempt in range(1, 9):
        try:
            resp = requests.get(URL, params=params, timeout=60)
        except requests.RequestException:
            resp = None  # network blip: treat like a 502
        status = resp.status_code if resp is not None else 502
        try:
            body = resp.json() if resp is not None else {}
        except ValueError:
            body = {}  # the per-IP 429 and proxy error pages are HTML

        if status == 200 and body.get("success"):
            return {"asin": asin, "status": "ok", "data": body["data"]}
        if status == 404:  # already charged: record it, never retry
            return {"asin": asin, "status": "missing"}
        if status in (401, 403) and "credits" in body.get("message", ""):
            raise OutOfCredits(body["message"])
        if status in (502, 503):
            paid_retries += 1
            if paid_retries > MAX_PAID_RETRIES:
                break
        elif status != 429:
            return {"asin": asin, "status": "error", "code": status}

        try:
            retry_after = float(resp.headers.get("Retry-After") or 0) if resp is not None else 0.0
        except ValueError:
            retry_after = 0.0
        time.sleep(max(retry_after, min(2 ** (attempt - 1), 32) + random.random()))
    return {"asin": asin, "status": "error", "code": "retries exhausted"}


def done() -> set[str]:
    """ASINs settled in an earlier run - read once, not per ASIN."""
    settled = set()
    if os.path.exists(OUT):
        with open(OUT, encoding="utf-8") as handle:
            for line in handle:
                try:
                    row = json.loads(line)
                except ValueError:
                    continue  # a half-written last line after a hard kill
                if row.get("status") in ("ok", "missing"):
                    settled.add(row["asin"])
    return settled


def main(csv_path: str, marketplace: str) -> None:
    finished = done()
    todo = [a for a in load_asins(csv_path) if a not in finished]
    print(f"{len(todo)} ASINs to fetch: {len(todo)} credits, plus any retried 502/503")
    pool = ThreadPoolExecutor(WORKERS)
    try:
        futures = [pool.submit(fetch, a, marketplace) for a in todo]
        with open(OUT, "a", encoding="utf-8") as out:
            for future in as_completed(futures):
                out.write(json.dumps(future.result()) + "\n")
                out.flush()
    finally:
        # Ctrl-C, a crash or OutOfCredits: drop the queue instead of paying for
        # ASINs that would never be written. Only in-flight requests finish.
        pool.shutdown(wait=True, cancel_futures=True)


if __name__ == "__main__":
    main(sys.argv[1], sys.argv[2])

Run it, stop it, run it again - finished ASINs are skipped

export SELLERMAGNET_API_KEY="your-key-here"
python bulk_lookup.py asins.csv ATVPDKIKX0DER

# how far did we get?
grep -c '"status": "ok"' results.jsonl
grep -c '"status": "missing"' results.jsonl

Exporting the results to CSV

to_csv.py - flatten results.jsonl for a spreadsheet

import csv
import json

with open("results.jsonl", encoding="utf-8") as src, \
        open("results.csv", "w", newline="", encoding="utf-8") as dst:
    writer = csv.writer(dst)
    writer.writerow(["asin", "title", "buy_box_price", "currency", "rank", "category"])
    for line in src:
        row = json.loads(line)
        if row["status"] != "ok":
            continue
        p = row["data"]["productInfo"]
        box = p.get("buyBoxInfo") or {}
        rank = (p.get("bestsellerRanks") or {}).get("main_category") or {}
        writer.writerow([row["asin"], p.get("title"), box.get("price"),
                         box.get("currencyCode"), rank.get("rank"), rank.get("name")])

Bottom line: bulk lookups are cheap when every unique ASIN is fetched exactly once. Clean the input, skip finished work, cap concurrency, retry only transient errors and write down the misses. The product lookup page lists every field each request returns, any of the 23 marketplace IDs works as the second argument, and pricing shows what a run costs. A free account includes 500 credits.