Python数据API限流怎么配置

本文目录导读：

1. 使用第三方库（最常用）
2. 使用asyncio实现异步限流
3. 手动实现简单的限流器
4. 针对特定API的限流配置
5. 企业级方案：使用Redis实现分布式限流
6. 最佳实践建议
7. 选择建议

在Python中配置数据API限流,通常有以下几种主流方案，具体取决于你的使用场景：

使用第三方库（最常用）

a) requests-ratelimiter（推荐）

from requests_ratelimiter import LimiterSession
import requests
# 配置每秒最多2个请求
session = LimiterSession(per_second=2)
for i in range(10):
    response = session.get('https://api.example.com/data')
    print(response.status_code)

b) ratelimit 装饰器

from ratelimit import limits, sleep_and_retry
# 每分钟最多60次调用
@sleep_and_retry
@limits(calls=60, period=60)
def call_api():
    response = requests.get('https://api.example.com/data')
    return response.json()

c) pyrate-limiter

from pyrate_limiter import Duration, RequestRate, Limiter
# 配置：每分钟60个请求
rate = RequestRate(60, Duration.MINUTE)
limiter = Limiter(rate)
@limiter.ratelimit("my_api", delay=True)
def call_api():
    response = requests.get('https://api.example.com/data')
    return response.json()

使用asyncio实现异步限流

import asyncio
import aiohttp
from asyncio_throttle import Throttler
async def main():
    throttler = Throttler(rate_limit=5)  # 每秒5次
    async with aiohttp.ClientSession() as session:
        for i in range(20):
            async with throttler:
                async with session.get('https://api.example.com/data') as response:
                    data = await response.json()
                    print(f"请求 {i}: {data['id']}")
asyncio.run(main())

手动实现简单的限流器

import time
from collections import deque
class RateLimiter:
    def __init__(self, max_calls, period=1):
        self.max_calls = max_calls
        self.period = period
        self.calls = deque()
    def wait_if_needed(self):
        now = time.time()
        # 清除过期的记录
        while self.calls and self.calls[0] < now - self.period:
            self.calls.popleft()
        if len(self.calls) >= self.max_calls:
            sleep_time = self.calls[0] + self.period - now
            if sleep_time > 0:
                print(f"等待 {sleep_time:.2f} 秒...")
                time.sleep(sleep_time)
        self.calls.append(time.time())
# 使用示例
limiter = RateLimiter(max_calls=5, period=1)  # 每秒最多5次
for i in range(10):
    limiter.wait_if_needed()
    response = requests.get('https://api.example.com/data')
    print(f"请求 {i} 成功")

针对特定API的限流配置

a) Twitter API

import tweepy
auth = tweepy.OAuth2BearerHandler("your_token")
api = tweepy.API(auth, wait_on_rate_limit=True, wait_on_rate_limit_notify=True)

b) OpenAI API

import openai
from ratelimit import limits, sleep_and_retry
openai.api_key = "your_key"
@sleep_and_retry
@limits(calls=60, period=60)
def chat_with_gpt(prompt):
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

企业级方案：使用Redis实现分布式限流

import redis
import time
class RedisRateLimiter:
    def __init__(self, redis_client, key_prefix="rate_limit"):
        self.redis = redis_client
        self.key_prefix = key_prefix
    def is_allowed(self, user_id, max_calls, period=60):
        key = f"{self.key_prefix}:{user_id}"
        current = self.redis.get(key)
        if current is None:
            self.redis.setex(key, period, 1)
            return True
        if int(current) >= max_calls:
            return False
        self.redis.incr(key)
        return True
# 使用
r = redis.Redis(host='localhost', port=6379, db=0)
limiter = RedisRateLimiter(r)
if limiter.is_allowed("user_123", max_calls=30, period=60):
    response = requests.get('https://api.example.com/data')
else:
    print("请求被限流，请等待")

最佳实践建议

import requests
from functools import wraps
from tenacity import retry, stop_after_attempt, wait_exponential
# 组合重试和限流
class RobustAPIClient:
    def __init__(self, base_url, rate=10, interval=1):
        self.base_url = base_url
        self.last_request = 0
        self.rate = rate
        self.interval = interval
    def rate_limit(self, func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - self.last_request
            wait_time = self.interval / self.rate - elapsed
            if wait_time > 0:
                time.sleep(wait_time)
            result = func(*args, **kwargs)
            self.last_request = time.time()
            return result
        return wrapper
    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
    @rate_limit
    def get_data(self, endpoint):
        response = requests.get(f"{self.base_url}/{endpoint}")
        response.raise_for_status()
        return response.json()
# 使用
client = RobustAPIClient("https://api.example.com", rate=10)
data = client.get_data("v1/data")

选择建议

关键点：

• 根据API文档仔细配置限流参数
• 始终添加适当的错误处理和重试机制
• 监控限流命中率,优化配置
• 考虑API可能存在突发流量的情况