如何通过Python的ratelimit实现API多时间维度请求速率限制?
多维度限流实现方案
没错,你需要通过叠加多个@limits装饰器来实现多维度限流——每个@limits装饰器对应一个时间窗口的限制规则,叠加后会同时校验所有维度的请求频率,只要触发任意一个限制就会自动等待直到符合条件。以下是修改后的完整代码:
import time import requests from ratelimit import limits, sleep_and_retry # 定义三个维度的限流参数 # 每秒最多25次请求 SEC_CALLS = 25 SEC_PERIOD = 1 # 每分钟最多250次请求 MIN_CALLS = 250 MIN_PERIOD = 60 # 每30分钟最多1000次请求 HALF_HOUR_CALLS = 1000 HALF_HOUR_PERIOD = 30 * 60 # 转换成秒 @sleep_and_retry @limits(calls=HALF_HOUR_CALLS, period=HALF_HOUR_PERIOD) @limits(calls=MIN_CALLS, period=MIN_PERIOD) @limits(calls=SEC_CALLS, period=SEC_PERIOD) def check_all_limits(): print(f'当前时间: {time.time()}') return # 假设ID是你的请求ID列表 for i in ID: check_all_limits() # 修复原代码的URL格式错误:用循环变量i而非列表ID url = 'https://xxxxxx{}'.format(i) headers = { 'Accept': 'application/json' } response = requests.get(url, headers=headers) # 检查响应状态 if response.status_code == 200: print(i) # print(response.json()) else: print(f"Error: {response.status_code} - {response.text}")
关键说明
- 装饰器顺序不影响限流逻辑,按时间窗口从大到小排列只是为了阅读方便
@sleep_and_retry只需加一次,它会处理所有@limits触发的限流异常,自动等待到符合规则后重试- 修正了原代码中
SEC_CALLS=10不符合需求的问题,同时修复了URL格式化的错误
当你每秒发起25次请求,10秒后触达每分钟250次的上限时,代码会自动暂停约50秒,直到分钟窗口的请求计数重置;后续也会自动遵守每30分钟1000次的限制规则。
内容的提问来源于stack exchange,提问作者pylearner
相关产品推荐
相关产品推荐

