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
相关产品推荐
相关产品推荐

