You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Amazon PA-API 5.0触发TooManyRequests错误,求解决方法

问题:调用Amazon Product Advertising API 5.0时持续收到429限流错误

我尝试通过Amazon Product Advertising API 5.0获取商品信息,但一直收到以下错误:

Error calling PA-API 5.0!
Status code: 429
Errors : {"__type":"com.amazon.paapi5#TooManyRequestsException","Errors":[{"Code":"TooManyRequests","Message":"The request was denied due to request throttling. Please verify the number of requests made per second to the Amazon Product Advertising API."}]}
Request ID: ee3f3ad0-e512-463f-b34b-395af81e347d

我已经按照官方文档指引操作,但问题仍未解决。我使用的是官方Python示例包中的sample_get_items_api.py脚本,代码如下:

from paapi5_python_sdk.api.default_api import DefaultApi
from paapi5_python_sdk.models.condition import Condition
from paapi5_python_sdk.models.get_items_request import GetItemsRequest
from paapi5_python_sdk.models.get_items_resource import GetItemsResource
from paapi5_python_sdk.models.partner_type import PartnerType
from paapi5_python_sdk.rest import ApiException


def parse_response(item_response_list):
    """
    The function parses GetItemsResponse and creates a dict of ASIN to Item object
    :param item_response_list: List of Items in GetItemsResponse
    :return: Dict of ASIN to Item object
    """
    mapped_response = {}
    for item in item_response_list:
        mapped_response[item.asin] = item
    return mapped_response


def get_items():
    """ Following are your credentials """
    """ Please add your access key here """
    access_key = "<YOUR ACCESS KEY>"

    """ Please add your secret key here """
    secret_key = "<YOUR SECRET KEY>"

    """ Please add your partner tag (store/tracking id) here """
    partner_tag = "<YOUR PARTNER TAG>"

    """ PAAPI host and region to which you want to send request """
    host = "webservices.amazon.com"
    region = "us-east-1"

    """ API declaration """
    default_api = DefaultApi(
        access_key=access_key, secret_key=secret_key, host=host, region=region
    )

    """ Request initialization"""

    """ Choose item id(s) """
    item_ids = ["059035342X", "B00X4WHP5E", "B00ZV9RDKK"]

    """ Choose resources you want from GetItemsResource enum """
    get_items_resource = [
        GetItemsResource.ITEMINFO_TITLE,
        GetItemsResource.OFFERS_LISTINGS_PRICE,
    ]

    """ Forming request """

    try:
        get_items_request = GetItemsRequest(
            partner_tag=partner_tag,
            partner_type=PartnerType.ASSOCIATES,
            marketplace="www.amazon.com",
            condition=Condition.NEW,
            item_ids=item_ids,
            resources=get_items_resource,
        )
    except ValueError as exception:
        print("Error in forming GetItemsRequest: ", exception)
        return

    try:
        """ Sending request """
        response = default_api.get_items(get_items_request)

        print("API called Successfully")
        print("Complete Response:", response)

        """ Parse response """
        if response.items_result is not None:
            print("Printing all item information in ItemsResult:")
            response_list = parse_response(response.items_result.items)
            for item_id in item_ids:
                print("Printing information about the item_id: ", item_id)
                if item_id in response_list:
                    item = response_list[item_id]
                    if item is not None:
                        if item.asin is not None:
                            print("ASIN: ", item.asin)
                        if item.detail_page_url is not None:
                            print("DetailPageURL: ", item.detail_page_url)
                        if (
                            item.item_info is not None
                            and item.item_info.title is not None
                            and item.item_info.title.display_value is not None
                        ):
                            print("Title: ", item.item_info.title.display_value)
                        if (
                            item.offers is not None
                            and item.offers.listings is not None
                            and item.offers.listings[0].price is not None
                            and item.offers.listings[0].price.display_amount is not None
                        ):
                            print(
                                "Buying Price: ",
                                item.offers.listings[0].price.display_amount,
                            )
                else:
                    print("Item not found, check errors")

        if response.errors is not None:
            print("\nPrinting Errors:\nPrinting First Error Object from list of Errors")
            print("Error code", response.errors[0].code)
            print("Error message", response.errors[0].message)

    except ApiException as exception:
        print("Error calling PA-API 5.0!")
        print("Status code:", exception.status)
        print("Errors :", exception.body)
        print("Request ID:", exception.headers["x-amzn-RequestId"])

    except TypeError as exception:
        print("TypeError :", exception)

    except ValueError as exception:
        print("ValueError :", exception)

    except Exception as exception:
        print("Exception :", exception)

请问如何解决这个限流问题?


解决方法
  • 控制请求频率:PA-API 5.0默认限制每秒1次请求,每日限额根据账户类型而定。先确认脚本是否在短时间内发送了过多请求,比如循环调用时未添加延迟。
  • 优化批量请求:利用API支持批量查询的特性,一次传入最多10个ASIN,减少请求次数,避免频繁发起单个请求。
  • 添加固定延迟:如果需要多次调用API,在每次请求之间添加至少1秒的延迟,示例代码:
    import time
    # 在请求发送后或循环中加入
    time.sleep(1)
    
  • 实现指数退避重试:针对429错误,添加重试逻辑,每次重试的等待时间翻倍(1秒、2秒、4秒...),避免持续触发限流。修改异常处理部分示例:
    import time
    max_retries = 3
    retry_count = 0
    while retry_count < max_retries:
        try:
            response = default_api.get_items(get_items_request)
            # 处理成功响应逻辑
            break
        except ApiException as exception:
            if exception.status == 429 and retry_count < max_retries:
                retry_count += 1
                wait_time = 2 ** retry_count
                print(f"限流错误,{wait_time}秒后重试...")
                time.sleep(wait_time)
            else:
                print("Error calling PA-API 5.0!")
                print("Status code:", exception.status)
                print("Errors :", exception.body)
                print("Request ID:", exception.headers["x-amzn-RequestId"])
                break
    
  • 确认账户限额:如果以上方法无效,检查PA-API账户是否有特殊限额,或是否因违规导致限额降低,可通过Amazon相关后台联系支持确认。

内容的提问来源于stack exchange,提问作者Mike Wilhelm

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.04 15:40:23