Python应用调用第三方API时处理速率限制的最佳实践问询

1. 手动实现速率限制 #

如果不想依赖第三方库，可手动实现两种主流速率控制策略：

时间窗口限制 #

针对固定时间窗口内的请求次数限制（比如每分钟100次），通过记录请求时间戳、清理过期记录来控制流量：

import time
from collections import deque

class RateLimiter:
    def __init__(self, max_requests, window_seconds):
        self.max_requests = max_requests
        self.window = window_seconds
        self.request_timestamps = deque()

    def wait(self):
        now = time.time()
        # 移除窗口外的历史请求时间戳
        while self.request_timestamps and now - self.request_timestamps[0] > self.window:
            self.request_timestamps.popleft()
        
        if len(self.request_timestamps) >= self.max_requests:
            # 计算需要等待的时长，直到窗口内请求数达标
            wait_time = self.window - (now - self.request_timestamps[0])
            if wait_time > 0:
                time.sleep(wait_time)
                # 等待后再次清理过期记录
                now = time.time()
                while self.request_timestamps and now - self.request_timestamps[0] > self.window:
                    self.request_timestamps.popleft()
        
        self.request_timestamps.append(now)

# 使用示例
limiter = RateLimiter(max_requests=100, window_seconds=60)
import requests

for _ in range(200):
    limiter.wait()
    response = requests.get("https://api.example.com/data")
    # 处理响应逻辑

令牌桶算法 #

适合允许突发请求的场景，按固定速率生成令牌，每次请求消耗1个令牌，无令牌则等待：

import time
import threading

class TokenBucket:
    def __init__(self, capacity, refill_rate):
        self.capacity = capacity  # 最大令牌存储量
        self.refill_rate = refill_rate  # 每秒生成令牌数
        self.tokens = capacity
        self.last_refill = time.time()
        self.lock = threading.Lock()

    def _refill(self):
        now = time.time()
        time_since_refill = now - self.last_refill
        new_tokens = time_since_refill * self.refill_rate
        self.tokens = min(self.capacity, self.tokens + new_tokens)
        self.last_refill = now

    def acquire(self):
        with self.lock:
            self._refill()
            if self.tokens >= 1:
                self.tokens -= 1
                return
            else:
                # 计算生成1个令牌所需等待时间
                wait_time = (1 - self.tokens) / self.refill_rate
                time.sleep(wait_time)
                self._refill()
                self.tokens -= 1

# 使用示例
bucket = TokenBucket(capacity=10, refill_rate=1)  # 每秒生成1个令牌，最多存10个
for _ in range(15):
    bucket.acquire()
    response = requests.get("https://api.example.com/data")
    # 处理响应逻辑

2. 推荐库与模式 #

专用速率限制库 #

• ratelimit：通过装饰器快速给请求函数添加速率限制，支持时间窗口策略：

  from ratelimit import limits, sleep_and_retry
  import requests
  
  # 限制每分钟最多100次请求
  @sleep_and_retry
  @limits(calls=100, period=60)
  def make_api_request(url):
      response = requests.get(url)
      response.raise_for_status()
      return response.json()
  
  # 批量调用示例
  url_list = ["https://api.example.com/data/1", "https://api.example.com/data/2"]
  for url in url_list:
      data = make_api_request(url)
  

• backoff：兼顾重试与速率控制，支持指数退避，可针对特定错误触发重试：

  import backoff
  import requests
  
  @backoff.on_exception(backoff.expo, requests.exceptions.HTTPError, max_tries=5)
  def make_api_request(url):
      response = requests.get(url)
      if response.status_code == 429:
          # 优先使用API返回的Retry-After头指定等待时间
          retry_after = int(response.headers.get("Retry-After", 10))
          time.sleep(retry_after)
          raise requests.exceptions.HTTPError("Rate limited", response=response)
      response.raise_for_status()
      return response.json()
  

请求库扩展 #

• requests-ratelimiter：直接给requests.Session绑定速率限制，无需修改原有请求代码：

  from requests_ratelimiter import LimiterSession
  
  # 配置每分钟最多100次请求
  session = LimiterSession(per_minute=100)
  response = session.get("https://api.example.com/data")
  

推荐模式 #

• 令牌桶/漏桶算法：令牌桶适合突发请求场景，漏桶适合平稳流量控制，根据API限制类型选择。
• 会话级统一控制：使用requests.Session管理所有请求，统一配置速率限制与重试逻辑，避免重复代码。

3. 优雅处理重试与退避策略 #

结合Retry-After的指数退避 #

遇到429错误时，优先遵循API返回的Retry-After头指定的等待时间，无该头则使用指数退避：

import time
import requests
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

class RateLimitError(Exception):
    pass

def handle_response(response):
    if response.status_code == 429:
        raise RateLimitError(f"Rate limited")
    response.raise_for_status()
    return response.json()

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=60),
    retry=retry_if_exception_type((RateLimitError, requests.exceptions.RequestException))
)
def make_api_request(url):
    response = requests.get(url)
    if response.status_code == 429:
        retry_after = int(response.headers.get("Retry-After", 0))
        if retry_after > 0:
            time.sleep(retry_after)
        raise RateLimitError()
    return handle_response(response)

重试注意事项 #

• 仅对可重试错误重试：比如429、5xx服务器错误，不要对400、401等客户端错误重试。
• 设置最大重试次数：避免无限循环消耗资源。
• 添加日志记录：记录速率限制触发、重试次数、等待时间，方便排查问题：

  import logging
  
  logging.basicConfig(level=logging.INFO)
  logger = logging.getLogger(__name__)
  
  @retry(...)
  def make_api_request(url):
      try:
          response = requests.get(url)
          if response.status_code == 429:
              retry_after = int(response.headers.get("Retry-After", 10))
              logger.info(f"触发速率限制，等待{retry_after}秒后重试")
              time.sleep(retry_after)
              raise RateLimitError()
          response.raise_for_status()
          return response.json()
      except Exception as e:
          logger.error(f"请求失败：{str(e)}，准备重试")
          raise
  

内容的提问来源于stack exchange，提问作者Online Water Bottle