Amazon API with asyncio and httpx: Concurrent Lookups in Python

httpx.AsyncClient with an asyncio.Semaphore looks up hundreds of Amazon ASINs concurrently from one Python process without opening more requests than the API allows per key. The server waits up to 20 seconds for a free concurrency slot before answering 429 with Retry-After: 2; that response is never billed, so the only cost of being too aggressive is a slower run, not wasted credits.

How many requests can run at once?

The API enforces a concurrency limit per key (40 at the same time) before it deducts a credit: a request over the limit waits up to 20 seconds for a slot, then gets a free 429 ("Max 40 concurrent requests per API key. Retry shortly.") with Retry-After: 2. A separate, also-free per-IP limit caps 600 requests per minute and 15,000 per hour. Running well under the per-key limit avoids both in practice.

Dependencies

pip install httpx
export SELLERMAGNET_API_KEY="your-key-here"

asyncio_api.py: async sibling of the quickstart's sellermagnet.py

"""Async sibling of the sellermagnet.py helper used by the Python quickstart
(https://sellermagnet-api.com/blog/amazon-product-api-python). Same contract
(ApiError, a 120s-timeout client, the same response-shape checks), implemented
with httpx.AsyncClient so many lookups can run concurrently from one process.
"""
import os

import httpx

API = "https://sellermagnet-api.com/api"


class ApiError(Exception):
    def __init__(self, status: int, message: str, retry_after: float = 0.0):
        super().__init__(f"{status}: {message}")
        self.status, self.retry_after = status, retry_after


def new_client() -> httpx.AsyncClient:
    return httpx.AsyncClient(
        base_url=API,
        headers={"X-Api-Key": os.environ["SELLERMAGNET_API_KEY"]},
        timeout=120.0,
    )


async def aget(client: httpx.AsyncClient, endpoint: str, **params) -> dict:
    resp = await client.get(f"/{endpoint}", params=params)
    try:
        body = resp.json()
    except ValueError:                      # e.g. an HTML page from a proxy
        body = {}
    if resp.status_code == 200 and body.get("success"):
        return body["data"]
    try:
        wait = float(resp.headers.get("Retry-After", 0))
    except ValueError:
        wait = 0.0
    raise ApiError(resp.status_code, body.get("message", "request failed"), wait)

async_retry.py: async sibling of the quickstart's retry.py

"""Async sibling of retry.py (same quickstart): retries only what a second
attempt can help with. 429 is the per-key concurrency limiter - free, so its
wait is capped on its own (MAX_WAITS) and never spends one of the attempts
budget below.
"""
import asyncio
import random

import httpx

from asyncio_api import ApiError, aget

RETRYABLE = {500, 502, 503}       # 429 is handled separately: free, but capped so a saturated key cannot loop forever
MAX_WAITS = 6                     # free 429 waits before giving up (about 2 minutes at the server's Retry-After)


async def aget_with_retry(client, endpoint: str, attempts: int = 4, **params) -> dict:
    internal_errors = 0
    waits = 0
    attempt = 1
    while True:
        try:
            return await aget(client, endpoint, **params)
        except ApiError as err:
            if err.status == 429:                               # free: wait Retry-After, never spend an attempt
                waits += 1
                if waits > MAX_WAITS:
                    raise
                await asyncio.sleep(max(err.retry_after, 1.0))
                continue
            internal_errors += err.status == 500
            if err.status not in RETRYABLE or attempt == attempts or internal_errors > 1:
                raise
            backoff = min(2 ** attempt, 30) + random.uniform(0, 1)
            await asyncio.sleep(max(backoff, err.retry_after))
            attempt += 1
        except (httpx.ConnectError, httpx.TimeoutException):
            if attempt == attempts:
                raise
            await asyncio.sleep(min(2 ** attempt, 30))
            attempt += 1

lookup_many.py: concurrent lookups with a bounded Semaphore

"""Concurrent Amazon product lookups with httpx.AsyncClient and a Semaphore
sized under the API's per-key concurrency limit (40 requests at the same
time per key; see https://sellermagnet-api.com/docs).
"""
import asyncio

from async_retry import aget_with_retry
from asyncio_api import ApiError, new_client

MARKETPLACE = "ATVPDKIKX0DER"


async def lookup_one(client, sem, asin, abort):
    if abort.is_set():                                # a rejected key/empty balance: stop spending requests
        return asin, None, "skipped"
    async with sem:                                   # never more than `concurrency` lookups in flight
        if abort.is_set():
            return asin, None, "skipped"
        try:
            p = (await aget_with_retry(client, "amazon-product-lookup",
                                        asin=asin, marketplaceId=MARKETPLACE))["productInfo"]
        except ApiError as err:
            if err.status in (401, 403):
                abort.set()                           # every remaining call would fail the same way
            return asin, None, err.status             # billed 404 included: a real answer, not an outage
        price = p["buyBoxInfo"].get("price")
        return asin, price if isinstance(price, (int, float)) else None, None  # "N/A" -> None, never 0


async def lookup_many(asins, concurrency: int = 10) -> dict:
    asins = list(dict.fromkeys(asins))                # dedupe: both copies would otherwise be billed
    sem = asyncio.Semaphore(concurrency)
    abort = asyncio.Event()
    async with new_client() as client:
        results = await asyncio.gather(
            *(lookup_one(client, sem, asin, abort) for asin in asins), return_exceptions=True)
    out = {}
    for asin, result in zip(asins, results):
        if isinstance(result, Exception):              # one bad ASIN never cancels the rest of the batch
            out[asin] = {"price": None, "error": type(result).__name__}
        else:
            _, price, status = result
            out[asin] = {"price": price, "error": status}
    return out


if __name__ == "__main__":
    asins = ["B0CL61F39H", "B0BVKV6SFV", "B0DEADBEEF"]
    print(asyncio.run(lookup_many(asins)))

Output of lookup_many.py (illustrative data)

{'B0CL61F39H': {'price': 444.99, 'error': None}, 'B0BVKV6SFV': {'price': 89.5, 'error': None}, 'B0DEADBEEF': {'price': None, 'error': 404}}

A newer asyncio.TaskGroup (Python 3.11+) is not used here because it cancels every sibling task the moment one fails; gather(return_exceptions=True) keeps every other lookup's result instead of throwing them all away.

How is this different from the threaded bulk lookup?

The bulk ASIN lookup guide and the Python quickstart use a thread pool: one OS thread per in-flight request, blocked on network I/O. asyncio with httpx.AsyncClient does the same waiting with far fewer OS threads, which starts to matter once a job runs thousands of ASINs from a small container. The API-side limits (the concurrency limit, the per-IP limiter, which status codes are billed) are identical either way: async is a client-side performance choice here, not a different contract with the API.

What should the per-request timeout be?

120 seconds, the same as the synchronous quickstart: the server allows a scrape that long, and the credit is deducted when the request starts, so a client that gives up early saves nothing (only a refunded 500 gives it back). httpx.AsyncClient(timeout=120.0) applies that to every request made through the shared client; a shorter client timeout only risks that outcome without actually limiting what the server does.

Bottom line: an asyncio Semaphore sized under the per-key concurrency limit, one shared httpx.AsyncClient, and a sleep-and-retry on the free 429 look up hundreds of ASINs quickly without wasting a single credit on a rate limit. Every limit named here is documented in the API docs.