云函数执行API调用触发429请求过多错误，本地运行正常求解决方案

本地环境执行API调用代码完全正常，但部署到Cloud Functions后收到429（Too Many Requests）错误，想知道是可以通过代码处理，还是账号存在问题？

附代码：

def get_visits(request):
    endpoint = "https://api.mercadolibre.com/items/visits?ids=" + str(i) + "&date_from=" + str(yesterday) + "T00:00:00.000-00:00&date_to=" + str(today) + "T00:00:00.000-00:00"
    payload={}
    headers = {'Authorization': 'Bearer ' + access_token}
    response = requests.get(endpoint, headers=headers, data=payload)
    if response.status_code != 200:    
        print('API response: ', response.text)
    else:
        return response.json()

原因排查与解决办法 #

一、先确认账号/API配额问题 #

• 查Mercado Libre API配额：登录开发者后台，查看你的应用是否达到调用频次上限。本地测试调用量小，部署后可能因触发频率高（比如定时批量任务）直接打满配额；另外Cloud Functions的共享出口IP可能被其他用户占用，导致你的请求被计入同一IP的集体配额，触发429。
• 核对权限与账号状态：确认你的access_token是否有效，账号是否存在API调用限制的处罚记录。

二、代码层面优化处理 #

1. 添加智能重试机制
   针对429错误自动重试，同时遵守API返回的Retry-After指引，避免无效重试：

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def get_visits(request):
    session = requests.Session()
    # 配置重试策略：针对429/5xx错误重试，间隔指数增长
    retry_strategy = Retry(
        total=3,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET"],
        backoff_factor=1
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)

    endpoint = f"https://api.mercadolibre.com/items/visits?ids={str(i)}&date_from={str(yesterday)}T00:00:00.000-00:00&date_to={str(today)}T00:00:00.000-00:00"
    headers = {'Authorization': 'Bearer ' + access_token}
    
    try:
        response = session.get(endpoint, headers=headers)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as err:
        if response.status_code == 429:
            retry_after = int(response.headers.get('Retry-After', 5))
            print(f"触发限流，{retry_after}秒后重试")
        print('API response: ', response.text)
        return None

2. 控制调用频率
   如果是批量处理多个item，在请求之间添加间隔，避免短时间内发送大量请求：

import time
# 批量处理示例
for item_id in item_ids:
    # 调用get_visits逻辑
    time.sleep(1)  # 每次请求间隔1秒，可根据API配额调整

3. 用批量请求减少调用次数
   查看Mercado Libre API文档，确认是否支持一次请求传入多个ids（用逗号分隔），这样能大幅减少请求总数：

# 假设item_ids是待查询的ID列表
ids_str = ",".join(map(str, item_ids))
endpoint = f"https://api.mercadolibre.com/items/visits?ids={ids_str}&date_from={str(yesterday)}T00:00:00.000-00:00&date_to={str(today)}T00:00:00.000-00:00"

三、Cloud Functions配置调整 #

• 调整触发频率：如果是定时触发，拉长执行间隔，避免短时间内重复执行函数。
• 使用静态出口IP：配置VPC连接器，让Cloud Functions使用固定IP请求API，避免共享IP的配额冲突。

内容的提问来源于stack exchange，提问作者curlydora