Python接口自动化中Token过期的9种解决方案

测试环境突然告警！公司核心业务的自动化测试用例批量失败，错误日志清一色显示“401 Unauthorized”。排查发现，所有接口请求的Token全部过期——这是每个接口自动化工程师都经历过的噩梦。Token过期看似简单，却可能导致整个测试链路瘫痪，尤其在持续集成环境中，一次Token失效就可能阻断版本发布流程。

作为深耕接口自动化领域8年的开发者，我整理了9种经过生产环境验证的Token过期解决方案。从简单粗暴的“暴力刷新”到企业级的“无感刷新”，从单机脚本到分布式系统，每种方案都附带完整Python代码实现和适用场景分析，帮你彻底解决Token管理难题。

固定时间刷新策略

适用场景：Token有效期固定且较长（如24小时）的内部系统，或对实时性要求不高的定时任务。

实现原理：在测试套件初始化时获取Token，并记录过期时间，每次请求前检查是否接近过期（预留10%缓冲时间），若接近则重新获取。这是最基础也最容易实现的方案，本质是通过时间预判主动规避过期风险。

import requests
import time
import logging
from datetime import datetime, timedelta

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class FixedTimeTokenManager:
    def __init__(self, login_url, credentials, buffer_ratio=0.1):
        self.login_url = login_url
        self.credentials = credentials  # 包含username和password的字典
        self.token = None
        self.expire_time = None
        self.buffer_ratio = buffer_ratio  # 过期缓冲比例

    def _get_new_token(self):
        """从登录接口获取新Token"""
        try:
            response = requests.post(self.login_url, json=self.credentials, timeout=10)
            response.raise_for_status()
            result = response.json()

            # 验证响应格式
            if 'access_token' not in result or 'expires_in' not in result:
                raise ValueError("登录响应缺少必要字段")

            self.token = result['access_token']
            # 计算过期时间，预留buffer_ratio比例的缓冲时间
            buffer_time = result['expires_in'] * self.buffer_ratio
            self.expire_time = time.time() + result['expires_in'] - buffer_time
            logger.info(f"Token刷新成功，有效期至: {datetime.fromtimestamp(self.expire_time)}")
            return self.token
        except Exception as e:
            logger.error(f"获取Token失败: {str(e)}", exc_info=True)
            raise  # 抛出异常让调用方处理

    def get_token(self):
        """获取可用Token，过期则自动刷新"""
        if not self.token or time.time() >= self.expire_time:
            logger.info("Token已过期或未初始化，尝试刷新")
            return self._get_new_token()
        return self.token

# 使用示例
if __name__ == "__main__":
    # 实际使用时替换为真实的登录URL和凭据
    TOKEN_MANAGER = FixedTimeTokenManager(
        login_url="https://api.example.com/login",
        credentials={"username": "test_user", "password": "test_pass"}
    )

    # 在测试用例中获取Token
    token = TOKEN_MANAGER.get_token()
    headers = {"Authorization": f"Bearer {token}"}
    # 使用headers发送请求...

优缺点分析：
✅ 优点：实现简单，资源消耗低，适用于大多数单机脚本场景；
✅ 优点：主动刷新机制避免请求失败，减少重试开销；
❌ 缺点：无法应对Token被提前吊销的情况；
❌ 缺点：分布式环境下多实例会重复刷新，浪费资源；
❌ 缺点：缓冲时间设置过短可能仍遇到过期，设置过长则频繁刷新。

企业级优化提议：生产环境中提议将buffer_ratio设置为0.1~0.2（即预留10%~20%的缓冲时间），并添加Token可用性校验接口，在关键业务前额外验证Token有效性。

响应拦截刷新策略

适用场景：Token过期时间不确定或服务端可能动态调整有效期的场景，如第三方API集成。

实现原理：利用请求拦截器捕获401响应，自动触发Token刷新并重新执行原请求。这种“被动防御”策略不需要预判过期时间，而是通过服务端反馈动态处理，特别适合无法准确获取过期时间的场景。

Python接口自动化中Token过期的9种解决方案

import requests
import logging
from functools import wraps

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class ResponseInterceptTokenManager:
    def __init__(self, login_url, credentials, max_retries=2):
        self.login_url = login_url
        self.credentials = credentials
        self.token = None
        self.session = requests.Session()
        self.max_retries = max_retries  # 最大重试次数
        # 为session添加请求/响应钩子
        self.session.hooks['response'].append(self._handle_response)

    def _get_new_token(self):
        """获取新Token"""
        try:
            response = requests.post(self.login_url, json=self.credentials, timeout=10)
            response.raise_for_status()
            result = response.json()
            self.token = result['access_token']
            logger.info("Token刷新成功")
            return self.token
        except Exception as e:
            logger.error(f"刷新Token失败: {str(e)}", exc_info=True)
            raise

    def _handle_response(self, response, kwargs):
        """响应拦截器，处理401错误"""
        # 只处理HTTP 401且未超过重试次数的请求
        if response.status_code == 401 and kwargs.get('retries', 0) < self.max_retries:
            logger.warning(f"收到401响应，尝试刷新Token (重试次数: {kwargs.get('retries', 0) + 1})")

            try:
                # 刷新Token
                self._get_new_token()

                # 重新构建请求
                request = response.request.copy()
                # 更新Authorization头
                request.headers['Authorization'] = f"Bearer {self.token}"

                # 增加重试标记，避免无限循环
                kwargs['retries'] = kwargs.get('retries', 0) + 1

                # 重新发送请求
                logger.info(f"重新发送请求: {request.url}")
                return self.session.send(request,kwargs)
            except Exception as e:
                logger.error(f"Token刷新后重发请求失败: {str(e)}")
                return response

        return response

    def get_session(self):
        """获取配置好拦截器的session"""
        if not self.token:
            self._get_new_token()
        # 设置初始Authorization头
        self.session.headers['Authorization'] = f"Bearer {self.token}"
        return self.session

# 使用示例
if __name__ == "__main__":
    # 初始化Token管理器
    TOKEN_MANAGER = ResponseInterceptTokenManager(
        login_url="https://api.example.com/login",
        credentials={"username": "test_user", "password": "test_pass"}
    )

    # 获取配置好的session
    session = TOKEN_MANAGER.get_session()

    # 正常发送请求，401错误会被自动处理
    try:
        response = session.get("https://api.example.com/protected_resource")
        response.raise_for_status()
        print(response.json())
    except requests.exceptions.HTTPError as e:
        # 处理其他HTTP错误
        logger.error(f"请求失败: {str(e)}")

优缺点分析：
✅ 优点：被动触发机制，无需预测过期时间，适应性更强；
✅ 优点：对业务代码侵入小，一次配置全局生效；
✅ 优点：能处理服务端主动吊销Token的情况；
❌ 缺点：请求失败后重试会增加响应时间；
❌ 缺点：极端情况下可能出现重试风暴（如刷新Token的接口也返回401）；
❌ 缺点：非401的Token无效场景（如403）无法处理。

企业级优化提议：实现时务必添加重试次数限制（如示例中的max_retries），并对刷新Token的请求单独处理，避免递归重试。对于可能返回403的场景，可在拦截器中同时处理401和403状态码。

预刷新机制

适用场景：高稳定性要求的自动化测试系统，如金融交易接口测试、核心业务监控等。

实现原理：启动独立线程定期检查Token有效期，在即将过期前（如剩余10%生命周期）自动刷新。这种“定时巡逻”机制结合了主动预判和后台刷新的优势，确保业务请求始终使用有效Token。

import requests
import time
import logging
import threading
from datetime import datetime
from threading import Event

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class PreRefreshTokenManager:
    def __init__(self, login_url, credentials, refresh_interval=60, pre_refresh_ratio=0.2):
        self.login_url = login_url
        self.credentials = credentials
        self.token = None
        self.expire_time = None
        self.refresh_interval = refresh_interval  # 检查间隔(秒)
        self.pre_refresh_ratio = pre_refresh_ratio  # 预刷新比例
        self._stop_event = Event()
        self._refresh_thread = None
        self._lock = threading.Lock()  # 线程安全锁

    def _get_new_token(self):
        """获取新Token（线程安全版）"""
        with self._lock:  # 确保同一时间只有一个线程刷新Token
            try:
                response = requests.post(self.login_url, json=self.credentials, timeout=10)
                response.raise_for_status()
                result = response.json()

                if 'access_token' not in result or 'expires_in' not in result:
                    raise ValueError("登录响应格式不正确")

                self.token = result['access_token']
                # 计算预刷新时间点（剩余生命周期达到pre_refresh_ratio比例时触发刷新）
                self.expire_time = time.time() + result['expires_in']
                logger.info(f"Token刷新成功，有效期至: {datetime.fromtimestamp(self.expire_time)}")
                return self.token
            except Exception as e:
                logger.error(f"获取Token失败: {str(e)}", exc_info=True)
                raise

    def _refresh_loop(self):
        """后台刷新循环"""
        logger.info("启动Token预刷新线程")
        while not self._stop_event.is_set():
            # 检查是否需要预刷新
            if self.token and self.expire_time:
                remaining_time = self.expire_time - time.time()
                total_lifetime = self.expire_time - (self.expire_time - remaining_time / (1 - self.pre_refresh_ratio))
                # 如果剩余时间小于预刷新比例，触发刷新
                if remaining_time / total_lifetime <= self.pre_refresh_ratio:
                    logger.info(f"Token即将过期（剩余时间: {remaining_time:.1f}秒），触发预刷新")
                    try:
                        self._get_new_token()
                    except Exception as e:
                        logger.error(f"预刷新Token失败: {str(e)}")

            # 等待指定间隔或直到停止事件被触发
            self._stop_event.wait(self.refresh_interval)
        logger.info("Token预刷新线程已停止")

    def start(self):
        """启动预刷新线程"""
        if not self._refresh_thread or not self._refresh_thread.is_alive():
            self._stop_event.clear()
            self._refresh_thread = threading.Thread(target=self._refresh_loop, daemon=True)
            self._refresh_thread.start()
            # 初始获取Token
            if not self.token:
                self._get_new_token()

    def stop(self):
        """停止预刷新线程"""
        self._stop_event.set()
        if self._refresh_thread and self._refresh_thread.is_alive():
            self._refresh_thread.join()

    def get_token(self):
        """获取当前Token"""
        if not self.token:
            raise RuntimeError("Token未初始化，请先调用start()")
        return self.token

# 使用示例
if __name__ == "__main__":
    TOKEN_MANAGER = PreRefreshTokenManager(
        login_url="https://api.example.com/login",
        credentials={"username": "test_user", "password": "test_pass"},
        refresh_interval=30,  # 每30秒检查一次
        pre_refresh_ratio=0.2  # 剩余20%生命周期时预刷新
    )

    try:
        # 启动预刷新线程
        TOKEN_MANAGER.start()

        # 在测试过程中获取Token
        while True:
            token = TOKEN_MANAGER.get_token()
            logger.info(f"当前Token: {token[:10]}...")
            time.sleep(10)  # 模拟测试间隔

    except KeyboardInterrupt:
        logger.info("程序被中断")
    finally:
        # 停止线程
        TOKEN_MANAGER.stop()

优缺点分析：
✅ 优点：业务请求无感知，始终使用最新Token；
✅ 优点：单线程后台刷新，资源消耗可控；
✅ 优点：结合了主动预判和后台处理的优势；
❌ 缺点：实现复杂度高于前两种方案；
❌ 缺点：多实例部署仍存在重复刷新问题；
❌ 缺点：线程管理增加系统复杂度，异常处理难度大。

企业级优化提议：在分布式系统中，可结合Redis分布式锁确保同一时间只有一个实例执行刷新；对于关键业务，提议实现Token状态监控告警，当连续刷新失败时及时通知管理员。

OAuth2.0自动刷新机制

适用场景：采用OAuth2.0认证框架的系统，如第三方API集成（Google、GitHub、企业微信等）。

实现原理：利用OAuth2.0协议定义的刷新令牌（Refresh Token）机制，当访问令牌（Access Token）过期时，使用刷新令牌获取新的访问令牌，避免用户重新登录。这种方案符合标准协议，兼容性好，安全性高。

Python接口自动化中Token过期的9种解决方案

import requests
import time
import logging
import json
from datetime import datetime
from typing import Optional, Dict, Any

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class OAuth2TokenManager:
    def __init__(self,
                 token_url: str,
                 client_id: str,
                 client_secret: str,
                 refresh_token: Optional[str] = None,
                 token_storage_path: Optional[str] = None):
        self.token_url = token_url
        self.client_id = client_id
        self.client_secret = client_secret
        self.refresh_token = refresh_token
        self.token_storage_path = token_storage_path  # Token持久化路径
        self.access_token = None
        self.expire_time = None
        # 尝试从存储加载Token
        self._load_tokens()

    def _request_token(self, grant_type: str, kwargs) -> Dict[str, Any]:
        """发送令牌请求"""
        payload = {
            "grant_type": grant_type,
            "client_id": self.client_id,
            "client_secret": self.client_secret, kwargs
        }

        try:
            response = requests.post(self.token_url, data=payload, timeout=10)
            response.raise_for_status()
            token_data = response.json()

            # 验证响应
            if 'access_token' not in token_data:
                raise ValueError("响应缺少access_token字段")

            # 更新刷新令牌（如果返回了新的）
            if 'refresh_token' in token_data:
                self.refresh_token = token_data['refresh_token']

            # 计算过期时间
            self.access_token = token_data['access_token']
            self.expire_time = time.time() + token_data.get('expires_in', 3600)  # 默认1小时有效期

            logger.info(f"获取Token成功，有效期至: {datetime.fromtimestamp(self.expire_time)}")
            # 持久化存储
            self._save_tokens()
            return token_data
        except Exception as e:
            logger.error(f"Token请求失败: {str(e)}", exc_info=True)
            raise

    def get_access_token(self) -> str:
        """获取访问令牌，过期则自动刷新"""
        if self.access_token and time.time() < self.expire_time:
            return self.access_token

        # 需要刷新Token
        if not self.refresh_token:
            raise RuntimeError("没有可用的refresh_token，无法刷新Token")

        logger.info("Access Token已过期，使用refresh_token刷新")
        self._request_token(grant_type="refresh_token", refresh_token=self.refresh_token)
        return self.access_token

    def _save_tokens(self) -> None:
        """持久化存储Token"""
        if self.token_storage_path and self.access_token and self.refresh_token:
            try:
                with open(self.token_storage_path, 'w') as f:
                    json.dump({
                        "access_token": self.access_token,
                        "refresh_token": self.refresh_token,
                        "expire_time": self.expire_time
                    }, f)
                logger.info(f"Token已保存到: {self.token_storage_path}")
            except Exception as e:
                logger.warning(f"保存Token到文件失败: {str(e)}")

    def _load_tokens(self) -> bool:
        """从存储加载Token"""
        if self.token_storage_path and os.path.exists(self.token_storage_path):
            try:
                with open(self.token_storage_path, 'r') as f:
                    token_data = json.load(f)
                    self.access_token = token_data.get('access_token')
                    self.refresh_token = token_data.get('refresh_token')
                    self.expire_time = token_data.get('expire_time')
                    logger.info(f"从文件加载Token，有效期至: {datetime.fromtimestamp(self.expire_time)}"
                                if self.expire_time else "从文件加载Token成功")
                    return True
            except Exception as e:
                logger.warning(f"从文件加载Token失败: {str(e)}")
        return False

    def authorize(self, code: str, redirect_uri: str) -> Dict[str, Any]:
        """使用授权码获取初始Token（首次认证用）"""
        return self._request_token(
            grant_type="authorization_code",
            code=code,
            redirect_uri=redirect_uri
        )

# 使用示例
if __name__ == "__main__":
    # 初始化OAuth2管理器
    OAUTH_MANAGER = OAuth2TokenManager(
        token_url="https://api.example.com/oauth/token",
        client_id="your_client_id",
        client_secret="your_client_secret",
        token_storage_path="oauth_tokens.json"  # Token持久化文件
    )

    # 如果是首次使用，需要先通过授权码获取初始Token
    # OAUTH_MANAGER.authorize(code="authorization_code", redirect_uri="https://yourapp.com/callback")

    # 在测试中获取访问令牌
    access_token = OAUTH_MANAGER.get_access_token()
    headers = {"Authorization": f"Bearer {access_token}"}
    # 使用headers发送请求...

优缺点分析：
✅ 优点：符合OAuth2.0标准，安全性高，适合第三方API集成；
✅ 优点：支持持久化存储，应用重启后无需重新授权；
✅ 优点：刷新令牌有效期长（一般30天以上），减少人工干预；
❌ 缺点：实现相对复杂，需要理解OAuth2.0协议；
❌ 缺点：刷新令牌一旦泄露风险较高；
❌ 缺点：部分服务提供商限制刷新令牌使用次数。

安全最佳实践：生产环境中应加密存储client_secret和refresh_token，可使用环境变量或加密配置文件；提议定期轮换client_secret，并监控异常的Token刷新行为。

JWT验证刷新

适用场景：使用JWT（JSON Web Token）作为认证令牌的系统，如前后端分离架构、微服务架构。

实现原理：JWT令牌本身包含过期时间（exp字段），通过解码令牌 payload 可直接获取过期时间，无需额外请求。这种方案省去了查询令牌状态的开销，特别适合分布式系统。

import requests
import time
import logging
import jwt
from datetime import datetime
from typing import Optional, Dict, Any

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class JWTTokenManager:
    def __init__(self,
                 login_url: str,
                 credentials: Dict[str, str],
                 public_key: Optional[str] = None,
                 algorithm: str = "HS256"):
        self.login_url = login_url
        self.credentials = credentials
        self.public_key = public_key  # 用于验证JWT签名的公钥
        self.algorithm = algorithm    # JWT签名算法
        self.token = None
        self.expire_time = None

    def _decode_token(self, token: str) -> Dict[str, Any]:
        """解码JWT令牌并验证签名"""
        try:
            # 如果提供了公钥，则验证签名
            if self.public_key:
                return jwt.decode(
                    token,
                    key=self.public_key,
                    algorithms=[self.algorithm],
                    options={"verify_exp": True}  # 验证过期时间
                )
            else:
                # 不验证签名，仅解码payload（用于无法获取公钥的场景）
                return jwt.decode(
                    token,
                    options={"verify_signature": False, "verify_exp": True}
                )
        except jwt.ExpiredSignatureError:
            raise ValueError("JWT令牌已过期")
        except jwt.InvalidTokenError as e:
            raise ValueError(f"无效的JWT令牌: {str(e)}")

    def _get_new_token(self) -> str:
        """从登录接口获取新JWT令牌"""
        try:
            response = requests.post(self.login_url, json=self.credentials, timeout=10)
            response.raise_for_status()
            result = response.json()

            if 'access_token' not in result:
                raise ValueError("登录响应缺少access_token字段")

            # 解码令牌获取过期时间
            payload = self._decode_token(result['access_token'])
            self.token = result['access_token']
            self.expire_time = payload['exp']  # JWT的exp字段是标准的过期时间戳
            logger.info(f"获取JWT令牌成功，有效期至: {datetime.fromtimestamp(self.expire_time)}")
            return self.token
        except Exception as e:
            logger.error(f"获取JWT令牌失败: {str(e)}", exc_info=True)
            raise

    def get_token(self) -> str:
        """获取可用JWT令牌，过期则刷新"""
        if self.token:
            try:
                # 尝试解码令牌验证有效性
                self._decode_token(self.token)
                return self.token
            except ValueError as e:
                logger.warning(f"JWT令牌无效: {str(e)}")

        # 令牌无效或已过期，重新获取
        logger.info("JWT令牌已过期或无效，重新获取")
        return self._get_new_token()

# 使用示例
if __name__ == "__main__":
    # 初始化JWT管理器
    # 如果服务端使用公钥签名JWT，需要提供public_key参数
    JWT_MANAGER = JWTTokenManager(
        login_url="https://api.example.com/login",
        credentials={"username": "test_user", "password": "test_pass"},
        # public_key=open("public_key.pem").read(),  # 实际使用时加载公钥
        # algorithm="RS256"  # 根据服务端使用的算法调整
    )

    # 获取JWT令牌
    token = JWT_MANAGER.get_token()
    headers = {"Authorization": f"Bearer {token}"}

    # 发送请求...
    response = requests.get("https://api.example.com/protected", headers=headers)
    response.raise_for_status()
    print(response.json())

优缺点分析：
✅ 优点：无需查询服务端即可验证令牌有效性，性能高；
✅ 优点：令牌自包含所有必要信息，适合分布式系统；
✅ 优点：支持多种加密算法（HS256、RS256等），安全性可控；
❌ 缺点：令牌一旦签发无法主动吊销（除非维护黑名单）；
❌ 缺点：需要处理密钥管理和轮换；
❌ 缺点：payload 解码不验证签名时有安全风险。

性能优化提议：对于高频请求场景，可缓存已解码的JWT payload，避免重复解码开销；使用非对称加密算法（如RS256）时，可预加载公钥并复用，减少I/O操作。

多Token策略

适用场景：高可用要求的关键业务系统，如支付系统、交易系统，需要最大限度减少认证失败风险。

实现原理：同时维护多个令牌（一般2~3个），主令牌过期时无缝切换到备用令牌，确保业务连续性。这种“冗余备份”机制能有效应对单个令牌刷新失败的情况。

import requests
import time
import logging
import random
from datetime import datetime
from typing import List, Dict, Optional

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class TokenInfo:
    """Token信息封装类"""
    def __init__(self, token: str, expire_time: float):
        self.token = token
        self.expire_time = expire_time

    def is_valid(self, buffer_time: float = 10) -> bool:
        """检查Token是否有效，预留buffer_time秒缓冲"""
        return time.time() + buffer_time < self.expire_time

    def __repr__(self) -> str:
        return f"TokenInfo(valid={self.is_valid()}, expire_time={datetime.fromtimestamp(self.expire_time)})"

class MultiTokenManager:
    def __init__(self,
                 login_url: str,
                 credentials_list: List[Dict[str, str]],
                 buffer_time: float = 10,
                 min_valid_tokens: int = 1):
        self.login_url = login_url
        self.credentials_list = credentials_list  # 多个用户凭据列表
        self.buffer_time = buffer_time  # 缓冲时间(秒)
        self.min_valid_tokens = min_valid_tokens  # 最小有效Token数量
        self.tokens: List[TokenInfo] = []
        self._lock = threading.Lock()  # 线程安全锁

    def _get_new_token(self, credentials: Dict[str, str]) -> TokenInfo:
        """使用指定凭据获取新Token"""
        try:
            response = requests.post(self.login_url, json=credentials, timeout=10)
            response.raise_for_status()
            result = response.json()

            if 'access_token' not in result or 'expires_in' not in result:
                raise ValueError("登录响应缺少必要字段")

            token = result['access_token']
            expire_time = time.time() + result['expires_in'] - self.buffer_time
            logger.info(f"获取Token成功（用户: {credentials.get('username')}），有效期至: {datetime.fromtimestamp(expire_time)}")
            return TokenInfo(token, expire_time)
        except Exception as e:
            logger.error(f"获取Token失败（用户: {credentials.get('username')}）: {str(e)}")
            raise

    def _refresh_all_tokens(self) -> None:
        """刷新所有Token"""
        with self._lock:
            new_tokens = []
            for credentials in self.credentials_list:
                try:
                    new_tokens.append(self._get_new_token(credentials))
                except Exception as e:
                    logger.error(f"刷新Token失败: {str(e)}")

            if len(new_tokens) < self.min_valid_tokens:
                raise RuntimeError(f"有效Token数量不足（要求: {self.min_valid_tokens}，实际: {len(new_tokens)}）")

            self.tokens = new_tokens
            logger.info(f"刷新所有Token完成，当前有效Token数量: {len(self.tokens)}")

    def get_token(self) -> str:
        """获取一个有效Token，优先选择剩余时间最长的"""
        with self._lock:
            # 过滤有效Token
            valid_tokens = [t for t in self.tokens if t.is_valid()]

            if len(valid_tokens) < self.min_valid_tokens:
                logger.warning(f"有效Token数量不足，触发全量刷新")
                self._refresh_all_tokens()
                valid_tokens = self.tokens  # 刷新后重新获取

            # 选择剩余时间最长的Token
            valid_tokens.sort(key=lambda x: x.expire_time, reverse=True)
            logger.info(f"选择Token（剩余时间: {valid_tokens[0].expire_time - time.time():.1f}秒）")
            return valid_tokens[0].token

    def initialize(self) -> None:
        """初始化Token池"""
        logger.info("初始化多Token池")
        self._refresh_all_tokens()

# 使用示例
if __name__ == "__main__":
    # 初始化多Token管理器
    # 实际使用时为不同用户配置不同凭据
    TOKEN_MANAGER = MultiTokenManager(
        login_url="https://api.example.com/login",
        credentials_list=[
            {"username": "test_user1", "password": "test_pass1"},
            {"username": "test_user2", "password": "test_pass2"},
            {"username": "test_user3", "password": "test_pass3"}
        ],
        min_valid_tokens=2  # 至少保持2个有效Token
    )

    # 初始化Token池
    TOKEN_MANAGER.initialize()

    # 在测试用例中获取Token
    token = TOKEN_MANAGER.get_token()
    headers = {"Authorization": f"Bearer {token}"}
    # 使用headers发送请求...

优缺点分析：
✅ 优点：冗余设计提高系统可用性，单个Token失效不影响业务；
✅ 优点：负载均衡效果，分散请求压力；
✅ 优点：支持平滑切换，适合需要无间断服务的场景；
❌ 缺点：需要多个用户账号，管理成本高；
❌ 缺点：资源消耗增加，每个Token都需要定期刷新；
❌ 缺点：不适合有用户上下文关联的业务。

最佳实践：提议为每个Token配置独立的用户账号，并限制单个账号的请求频率；在分布式系统中，可结合服务发现机制动态选择Token。

Redis分布式存储

适用场景：分布式接口自动化测试框架，多节点同时运行测试用例的场景。

实现原理：使用Redis聚焦存储Token，所有测试节点共享同一个Token，由一个节点负责刷新，其他节点被动读取。这种方案避免了分布式环境下的Token重复刷新问题。

import requests
import time
import logging
import redis
import threading
from datetime import datetime
from typing import Optional, Dict, Any

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class RedisTokenManager:
    def __init__(self,
                 login_url: str,
                 credentials: Dict[str, str],
                 redis_url: str,
                 redis_key: str = "api:automation:token",
                 refresh_interval: int = 60,
                 buffer_time: int = 60):
        self.login_url = login_url
        self.credentials = credentials
        self.redis_key = redis_key
        self.refresh_interval = refresh_interval  # 刷新检查间隔(秒)
        self.buffer_time = buffer_time  # 缓冲时间(秒)
        self._redis = redis.from_url(redis_url)
        self._lock = threading.Lock()
        self._refresh_thread = None
        self._stop_event = threading.Event()
        self._is_leader = False  # 是否为刷新领导者节点

    def _get_new_token(self) -> Dict[str, Any]:
        """获取新Token"""
        try:
            response = requests.post(self.login_url, json=self.credentials, timeout=10)
            response.raise_for_status()
            result = response.json()

            if 'access_token' not in result or 'expires_in' not in result:
                raise ValueError("登录响应缺少必要字段")

            # 计算过期时间，预留buffer_time缓冲
            expire_time = time.time() + result['expires_in'] - self.buffer_time
            token_data = {
                "access_token": result['access_token'],
                "expire_time": expire_time,
                "updated_at": time.time()
            }

            # 存储到Redis
            self._redis.setex(
                self.redis_key,
                int(result['expires_in']),  # Redis过期时间
                json.dumps(token_data)
            )

            logger.info(f"Token刷新成功，有效期至: {datetime.fromtimestamp(expire_time)}")
            return token_data
        except Exception as e:
            logger.error(f"获取新Token失败: {str(e)}", exc_info=True)
            raise

    def _refresh_leader_loop(self) -> None:
        """领导者节点刷新循环"""
        logger.info("成为Token刷新领导者节点，启动刷新循环")
        while not self._stop_event.is_set():
            # 检查Redis中的Token是否即将过期
            try:
                token_data_str = self._redis.get(self.redis_key)
                if token_data_str:
                    token_data = json.loads(token_data_str)
                    # 如果剩余时间小于刷新间隔，触发刷新
                    remaining_time = token_data['expire_time'] - time.time()
                    if remaining_time <= self.refresh_interval:
                        logger.info(f"Token即将过期（剩余时间: {remaining_time:.1f}秒），触发刷新")
                        self._get_new_token()
                else:
                    logger.warning("Redis中未找到Token，触发刷新")
                    self._get_new_token()
            except Exception as e:
                logger.error(f"刷新循环出错: {str(e)}")

            # 等待刷新间隔
            self._stop_event.wait(self.refresh_interval)
        logger.info("Token刷新领导者节点循环已停止")

    def _elect_leader(self) -> bool:
        """领导者选举，使用Redis的SETNX实现"""
        leader_key = f"{self.redis_key}:leader"
        # 设置领导者锁，有效期为刷新间隔的2倍
        if self._redis.set(leader_key, "1", nx=True, ex=self.refresh_interval * 2):
            logger.info("成功当选Token刷新领导者")
            return True
        return False

    def start(self) -> None:
        """启动Token管理服务"""
        # 尝试选举领导者
        if self._elect_leader():
            self._is_leader = True
            self._refresh_thread = threading.Thread(target=self._refresh_leader_loop, daemon=True)
            self._refresh_thread.start()
        else:
            logger.info("未当选领导者，使用其他节点刷新的Token")

        # 确保Redis中有可用Token
        if not self.get_token():
            with self._lock:
                # 如果不是领导者但Token不存在，强制获取
                if not self._is_leader:
                    logger.warning("Redis中无可用Token，强制获取")
                    self._get_new_token()

    def stop(self) -> None:
        """停止服务"""
        self._stop_event.set()
        if self._refresh_thread and self._refresh_thread.is_alive():
            self._refresh_thread.join()

    def get_token(self) -> str:
        """从Redis获取Token"""
        try:
            token_data_str = self._redis.get(self.redis_key)
            if not token_data_str:
                return None

            token_data = json.loads(token_data_str)
            # 检查是否过期
            if time.time() >= token_data['expire_time']:
                logger.warning("Redis中的Token已过期")
                return None

            return token_data['access_token']
        except Exception as e:
            logger.error(f"从Redis获取Token失败: {str(e)}")
            return None

# 使用示例
if __name__ == "__main__":
    # 初始化Redis Token管理器
    TOKEN_MANAGER = RedisTokenManager(
        login_url="https://api.example.com/login",
        credentials={"username": "test_user", "password": "test_pass"},
        redis_url="redis://localhost:6379/0",  # Redis连接URL
        refresh_interval=60,  # 每分钟检查一次
        buffer_time=30  # 预留30秒缓冲
    )

    # 启动服务
    TOKEN_MANAGER.start()

    # 获取Token
    token = TOKEN_MANAGER.get_token()
    if token:
        headers = {"Authorization": f"Bearer {token}"}
        # 发送请求...
    else:
        logger.error("无法获取有效Token")

优缺点分析：
✅ 优点：分布式环境下仅一个节点刷新Token，减少资源消耗；
✅ 优点：Token聚焦存储，所有节点共享，一致性好；
✅ 优点：支持故障转移，领导者节点故障后自动重新选举；
❌ 缺点：依赖Redis服务，增加系统复杂度；
❌ 缺点：网络延迟可能导致Token更新不及时；
❌ 缺点：领导者选举存在短暂的竞争窗口。

分布式部署提议：Redis推荐使用主从复制+哨兵模式确保高可用；生产环境中提议将refresh_interval设置为Token有效期的1/5~1/10，平衡及时性和性能开销。

重试机制

适用场景：Token过期偶发且重试成本低的场景，如非核心业务接口测试、低频率请求场景。

实现原理：当请求因Token过期失败时，捕获特定异常并重试，重试前刷新Token。这种“故障后修复”的被动策略实现简单，适合对稳定性要求不高的场景。

import requests
import time
import logging
from functools import wraps
from typing import Callable, Any

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

def token_retry_decorator(max_retries: int = 2, backoff_factor: float = 0.3):
    """Token过期重试装饰器"""
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            last_exception = None
            # 重试循环
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except requests.exceptions.HTTPError as e:
                    last_exception = e
                    # 检查是否是401未授权错误
                    if e.response.status_code == 401 and attempt < max_retries:
                        # 检查是否有Token管理器参数
                        token_manager = kwargs.get('token_manager') or getattr(args[0], 'token_manager', None)
                        if not token_manager:
                            logger.warning("未找到Token管理器，无法刷新Token")
                            raise

                        # 计算退避时间
                        sleep_time = backoff_factor * (2 ** attempt)
                        logger.warning(f"请求失败（401未授权），尝试刷新Token并重试（第{attempt+1}次，退避: {sleep_time:.2f}秒）")

                        try:
                            # 刷新Token
                            token_manager.refresh_token()
                            # 退避等待
                            time.sleep(sleep_time)
                            continue
                        except Exception as refresh_e:
                            logger.error(f"Token刷新失败: {str(refresh_e)}")
                            raise
                    raise  # 非401错误直接抛出
                except Exception as e:
                    last_exception = e
                    if attempt < max_retries:
                        sleep_time = backoff_factor * (2 attempt)
                        logger.warning(f"请求异常，尝试重试（第{attempt+1}次，退避: {sleep_time:.2f}秒）: {str(e)}")
                        time.sleep(sleep_time)
                        continue
                    raise

            # 所有重试失败
            raise last_exception
        return wrapper
    return decorator

class SimpleTokenManager:
    """简单Token管理器"""
    def __init__(self, login_url: str, credentials: dict):
        self.login_url = login_url
        self.credentials = credentials
        self.token = None

    def refresh_token(self) -> str:
        """刷新Token"""
        try:
            response = requests.post(self.login_url, json=self.credentials, timeout=10)
            response.raise_for_status()
            self.token = response.json()['access_token']
            logger.info("Token刷新成功")
            return self.token
        except Exception as e:
            logger.error(f"Token刷新失败: {str(e)}", exc_info=True)
            raise

    def get_token(self) -> str:
        """获取Token"""
        if not self.token:
            self.refresh_token()
        return self.token

# 使用示例
class ApiClient:
    def __init__(self, token_manager):
        self.token_manager = token_manager

    @token_retry_decorator(max_retries=2)
    def get_protected_resource(self):
        """调用需要认证的接口"""
        token = self.token_manager.get_token()
        headers = {"Authorization": f"Bearer {token}"}
        response = requests.get(
            "https://api.example.com/protected",
            headers=headers,
            timeout=10
        )
        response.raise_for_status()
        return response.json()

# 初始化
TOKEN_MANAGER = SimpleTokenManager(
    login_url="https://api.example.com/login",
    credentials={"username": "test_user", "password": "test_pass"}
)
CLIENT = ApiClient(token_manager=TOKEN_MANAGER)

# 调用接口（会自动处理重试）
try:
    data = CLIENT.get_protected_resource()
    print(data)
except Exception as e:
    logger.error(f"调用接口失败: {str(e)}")

优缺点分析：
✅ 优点：实现简单，对现有代码侵入小（通过装饰器）；
✅ 优点：资源消耗低，仅在失败时触发刷新；
✅ 优点：支持退避策略，减少服务端压力；
❌ 缺点：请求失败后重试会增加响应时间；
❌ 缺点：不适合写操作，可能导致重复提交；
❌ 缺点：频繁过期会严重影响性能。

最佳实践：结合幂等设计使用，确保重试不会导致副作用；设置合理的max_retries（提议23次）和backoff_factor（提议0.31秒）；关键业务接口提议结合主动刷新机制。

无感刷新

适用场景：对用户体验要求高的自动化系统，如持续集成流水线、监控告警系统。

实现原理：维护一个长期有效的刷新令牌（Refresh Token），当访问令牌过期时，使用刷新令牌静默获取新的访问令牌，整个过程对业务逻辑透明，用户无感知。

import requests
import time
import logging
import json
import os
from datetime import datetime
from typing import Optional, Dict, Any

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class SilentRefreshTokenManager:
    def __init__(self,
                 token_url: str,
                 client_id: str,
                 client_secret: str,
                 refresh_token: Optional[str] = None,
                 token_storage_path: str = "tokens.json"):
        self.token_url = token_url
        self.client_id = client_id
        self.client_secret = client_secret
        self.refresh_token = refresh_token
        self.token_storage_path = token_storage_path
        self.access_token = None
        self.expire_time = None
        # 尝试从存储加载
        self._load_tokens()

    def _save_tokens(self) -> None:
        """保存令牌到文件"""
        if self.access_token and self.refresh_token and self.expire_time:
            try:
                with open(self.token_storage_path, 'w') as f:
                    json.dump({
                        "access_token": self.access_token,
                        "refresh_token": self.refresh_token,
                        "expire_time": self.expire_time
                    }, f)
                logger.info(f"令牌已保存到: {self.token_storage_path}")
            except Exception as e:
                logger.warning(f"保存令牌失败: {str(e)}")

    def _load_tokens(self) -> bool:
        """从文件加载令牌"""
        if os.path.exists(self.token_storage_path):
            try:
                with open(self.token_storage_path, 'r') as f:
                    data = json.load(f)
                    self.access_token = data.get('access_token')
                    self.refresh_token = data.get('refresh_token')
                    self.expire_time = data.get('expire_time')
                    logger.info(f"从文件加载令牌，有效期至: {datetime.fromtimestamp(self.expire_time)}"
                                if self.expire_time else "从文件加载令牌成功")
                    return True
            except Exception as e:
                logger.warning(f"加载令牌失败: {str(e)}")
        return False

    def _request_token(self, grant_type: str,kwargs) -> Dict[str, Any]:
        """请求令牌"""
        payload = {
            "grant_type": grant_type,
            "client_id": self.client_id,
            "client_secret": self.client_secret,** kwargs
        }

        try:
            response = requests.post(self.token_url, data=payload, timeout=10)
            response.raise_for_status()
            result = response.json()

            # 检查响应
            if 'access_token' not in result:
                raise ValueError("响应缺少access_token")

            # 更新令牌
            self.access_token = result['access_token']
            self.expire_time = time.time() + result.get('expires_in', 3600)  # 默认1小时

            # 如果返回了新的刷新令牌则更新
            if 'refresh_token' in result:
                self.refresh_token = result['refresh_token']

            logger.info(f"获取令牌成功，有效期至: {datetime.fromtimestamp(self.expire_time)}")
            self._save_tokens()
            return result
        except Exception as e:
            logger.error(f"请求令牌失败: {str(e)}", exc_info=True)
            raise

    def get_access_token(self) -> str:
        """获取访问令牌，无感刷新"""
        # 检查当前令牌是否有效
        if self.access_token and time.time() < self.expire_time:
            return self.access_token

        # 需要刷新令牌
        if not self.refresh_token:
            raise RuntimeError("没有可用的refresh_token，请先进行授权")

        logger.info("访问令牌已过期，使用refresh_token进行无感刷新")
        self._request_token(grant_type="refresh_token", refresh_token=self.refresh_token)
        return self.access_token

    def authorize(self, code: str, redirect_uri: str) -> Dict[str, Any]:
        """使用授权码获取初始令牌（首次使用）"""
        return self._request_token(
            grant_type="authorization_code",
            code=code,
            redirect_uri=redirect_uri
        )

# 使用示例
if __name__ == "__main__":
    # 初始化令牌管理器
    TOKEN_MANAGER = SilentRefreshTokenManager(
        token_url="https://api.example.com/oauth/token",
        client_id="your_client_id",
        client_secret="your_client_secret",
        token_storage_path="auth_tokens.json"
    )

    # 首次使用需要授权码
    # 如果没有保存的refresh_token，需要调用authorize方法
    # TOKEN_MANAGER.authorize(code="your_authorization_code", redirect_uri="https://yourapp.com/callback")

    # 获取访问令牌（自动无感刷新）
    token = TOKEN_MANAGER.get_access_token()
    headers = {"Authorization": f"Bearer {token}"}
    # 使用headers发送请求...