带Token认证的多API请求优化：过期检查与QPS控制咨询

我来帮你拆解这两个问题，给出落地性强的解决方案：

问题1：Token过期检查与自动刷新的最优实现 #

最优的方式是把Token的管理逻辑封装成一个独立的类，让它负责Token的获取、过期检查和自动刷新，避免在业务代码里重复写判断逻辑。核心思路是：

• 保存当前有效的auth_token和对应的过期时间戳auth_expires
• 每次需要使用Token前，先检查当前Token是否已过期（或者预留一小段缓冲时间，比如提前30秒刷新，避免网络延迟导致过期）
• 如果过期/即将过期，重新调用POST /api/authorize获取新Token

给你一个示例实现：

import requests
import time

class TokenManager:
    def __init__(self, version, client_id, client_secret):
        self.version = version
        self.client_id = client_id
        self.client_secret = client_secret
        self.auth_token = None
        self.auth_expires = 0  # 过期时间戳，初始为0表示无有效Token

    def _fetch_new_token(self):
        """内部方法：调用授权接口获取新Token"""
        response = requests.post(
            "/api/authorize",
            data={
                "version": self.version,
                "client_id": self.client_id,
                "client_secret": self.client_secret
            }
        )
        response.raise_for_status()  # 处理请求失败的情况
        token_data = response.json()
        self.auth_token = token_data["auth_token"]
        # 把过期时间转换成时间戳（假设auth_expires是秒数，比如返回的是"3600"表示1小时后过期）
        self.auth_expires = time.time() + int(token_data["auth_expires"])

    def get_valid_token(self):
        """对外方法：返回有效的Token，自动刷新过期的"""
        # 预留30秒缓冲，避免刚好在请求时Token过期
        if self.auth_token is None or time.time() + 30 >= self.auth_expires:
            self._fetch_new_token()
        return self.auth_token

使用的时候，只需要初始化一次TokenManager，每次请求前调用get_valid_token()就能拿到有效的Token，不用自己管过期逻辑。

问题2：优化API调用模式（复用Session+QPS控制） #

当前每次请求创建新requests.Session的问题在于，会重复建立TCP连接，大大降低请求效率。同时加上QPS控制，可以避免触发API的限流机制。最优的模式是：

1. 复用单个requests.Session实例：Session会保持TCP连接池，减少握手开销
2. 整合RateLimiter：用限流工具控制每秒的请求次数，比如ratelimit库或者手动实现简单的限流逻辑

下面是整合了Token管理、Session复用和QPS控制的完整示例：

from ratelimit import limits, sleep_and_retry
import requests
import time

# 先定义Token管理类（和上面的TokenManager一致）
class TokenManager:
    def __init__(self, version, client_id, client_secret):
        self.version = version
        self.client_id = client_id
        self.client_secret = client_secret
        self.auth_token = None
        self.auth_expires = 0

    def _fetch_new_token(self):
        response = requests.post(
            "/api/authorize",
            data={
                "version": self.version,
                "client_id": self.client_id,
                "client_secret": self.client_secret
            }
        )
        response.raise_for_status()
        token_data = response.json()
        self.auth_token = token_data["auth_token"]
        self.auth_expires = time.time() + int(token_data["auth_expires"])

    def get_valid_token(self):
        if self.auth_token is None or time.time() + 30 >= self.auth_expires:
            self._fetch_new_token()
        return self.auth_token

# 定义API调用类，整合Session、Token和限流
class ProductAPIClient:
    def __init__(self, version, client_id, client_secret, qps_limit=5):
        self.token_manager = TokenManager(version, client_id, client_secret)
        self.session = requests.Session()  # 复用单个Session
        self.qps_limit = qps_limit
        # 用ratelimit库实现QPS控制，这里限制每秒最多5次请求
        self._get_product_data = self._wrap_rate_limiter(self._get_product_data)

    def _wrap_rate_limiter(self, func):
        """包装函数，添加QPS限流"""
        @sleep_and_retry
        @limits(calls=self.qps_limit, period=1)
        def wrapped(*args, **kwargs):
            return func(*args, **kwargs)
        return wrapped

    def _get_product_data(self, product_id):
        """内部方法：实际调用GetProductData接口"""
        token = self.token_manager.get_valid_token()
        headers = {"Authorization": f"Bearer {token}"}
        response = self.session.get(
            f"/api/products/{product_id}",
            headers=headers
        )
        response.raise_for_status()
        return response.json()

    def process_products(self, df):
        """处理DataFrame中的产品数据"""
        results = []
        for _, row in df.iterrows():
            product_id = row["product_id"]
            try:
                data = self._get_product_data(product_id)
                results.append(data)
            except Exception as e:
                print(f"处理产品{product_id}失败: {str(e)}")
                results.append(None)
        return results

关键优化点说明： #

• 复用Session：requests.Session会自动维护TCP连接池，多次请求复用同一个连接，大幅提升请求速度
• QPS控制：通过ratelimit库的@limits装饰器，轻松实现每秒请求次数限制，sleep_and_retry会在触发限流时自动等待并重试
• 解耦逻辑：Token管理、API调用、业务处理分开，代码更易维护和扩展

如果不想用第三方库，也可以手动实现简单的限流逻辑，比如用一个队列记录请求时间，每次请求前检查队列里的时间是否超过1秒窗口，不足的话等待。

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