Python循环调用Google API报429错误如何减速不中断程序运行

核心优化点 #

• 主动控速：每次正常请求完成后添加1秒固定等待，从源头降低请求频率，减少触发速率限制的概率
• 指数退避重试：替换原固定1秒延迟逻辑，429报错后重试等待时间逐次翻倍，最大等待60秒，兼顾重试效率和配额恢复需求
• 修复原有代码逻辑漏洞：原代码异常分支未正确获取响应状态码，修正后直接从请求异常对象中提取状态码判断
• 配额消耗监控：新增请求计数逻辑，累计请求量接近2万配额上限时主动终止，避免配额耗尽

优化后完整代码 #

import time
import requests

def connGoogleAPI(booksData):
    key = "**************************"
    books_list = []
    col = ['Title', 'Authors', 'published Date', 'Description','ISBN']
    books_list.append(col)
    # 配额上限预留10%缓冲，避免超量
    QUOTA_LIMIT = 18000
    request_count = 0

    with requests.Session() as session:
        for author in booksData:
            params = {"q": author, "key": key, "maxResults": 1}
            # 指数退避初始延迟
            retry_delay = 1
            max_retry_delay = 60
            res = None
            while True:
                # 达到配额上限直接终止
                if request_count >= QUOTA_LIMIT:
                    raise Exception("已接近配额上限，任务终止")
                try:
                    response = session.get("https://www.googleapis.com/books/v1/volumes", params=params)
                    request_count += 1
                    response.raise_for_status()
                    res = response.json()
                    break
                except requests.exceptions.HTTPError as e:
                    if e.response.status_code == 429:
                        if retry_delay > max_retry_delay:
                            raise Exception(f"作者 {author} 重试多次仍触发速率限制，跳过该条") from e
                        time.sleep(retry_delay)
                        retry_delay *= 2
                    else:
                        raise e
                except Exception as e:
                    raise e
            # 正常请求完成后添加固定间隔，主动降速
            time.sleep(1)
            books_list.append(lookup(res, author))
    return books_list

# 函数定义完成后再调用
author_List = ["J. K. Rowling", "mark twain","Emily Dickinson"] 
result = connGoogleAPI(author_List)

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