17370845950

本文介绍如何通过封装http请求方法而非使用装饰器，来优雅处理api调用中因token过期（http 401）触发的自动重试与令牌刷新逻辑，兼顾可维护性、可读性与面向对象设计原则。

在构建与需身份认证的REST API交互的Python客户端时，Token过期导致的 401 Unauthorized 是高频问题。虽然初看装饰器（decorator）似乎是“自动重试+刷新Token”的理想解法，但实际工程实践中，直接封装请求逻辑往往比强行套用装饰器更清晰、更可控、更符合职责分离原则。

如原始代码所示，每个业务方法（如 get_device）内嵌重复的 401 判断与重发逻辑，不仅冗余，还破坏了单一职责。而若强行设计一个通用装饰器（例如 @retry_on_401），将面临几个关键挑战：

• 装饰器需访问实例属性（如 self.session、self.get_token()），但标准函数装饰器无法天然感知 self；
• 若改用类装饰器或 functools.wraps + 闭包，会显著增加复杂度，且难以统一管理Token刷新状态（如避免并发重复刷新）；
• 装饰器通常作用于函数层面，而Token刷新本质是客户端会话（session）生命周期管理，属于类的内部状态行为，更适合在类内部抽象。

因此，推荐采用封装请求辅助方法的方案，如答案中所示：

import json
from typing import Any, Dict

class APIClient:
    def __init__(self, username: str, password: str):
        self.username = username
        self.password = password
        self.session = requests.Session()
        self._token = None

    def get_token(self) -> str:
        payload = json.dumps({"name": self.username, "password": self.password})
        r = self.session.post(
            BASE_URL + "authentication/signin",
            data=payload,
            verify=False,
        )
        if not r.ok:
            raise RuntimeError(f"Login failed: {r.status_code} {r.text}")
        response = r.json()
        if "token" not in response:
            raise RuntimeError("Token missing in login response")
        self._token = response["token"]
        return self._token

    def _authenticated_post(self, url: str, **kwargs) -> requests.Response:
        # 首次请求
        r = self.session.post(url, **kwargs)
        # 若未授权，刷新Token并重试
        if r.status_code == 401:
            self.session.headers.update({"X-AUTH-TOKEN": self.get_token()})
            r = self.session.post(url, **kwargs)
        return r

    def get_device(self, ip: str) -> Dict[str, Any]:
        data = json.dumps({"ipAddress": ip.strip()})
        r = self._authenticated_post(
            BASE_URL + "devices/filter",
            data=data
        )
        if not r.ok:
            raise RuntimeError(f"API request failed: {r.status_code} {r.text}")
        response = r.json()
        if "content" not in response:
            raise RuntimeError("Expected 'content' field not found")
        return response["content"][0]

✅ 优势说明：

• 语义明确：_authenticated_post 清晰表达了“带自动认证保障的POST”意图；
• 状态可控：Token缓存（self._token）与刷新逻辑集中管理，便于扩展（如添加过期时间校验、异步刷新、失败降级等）；
• 易于测试：可单独对 _authenticated_post 单元测试模拟 401 场景；
• 无侵入性：业务方法保持简洁，专注领域逻辑，不耦合认证细节。

⚠️ 注意事项：

• 生产环境应禁用 verify=False，改用可信CA证书或指定证书路径；
• 建议为 get_token() 添加简单防重入锁（如 threading.Lock），避免高并发下多次重复登录；
• 可进一步将 _authenticated_post 泛化为 _request(method, url, **kwargs)，统一支持 GET/PUT/DELETE 等方法；
• 如需全局拦截所有请求，可考虑 requests.Session 的 mount() + 自定义 HTTPAdapter，但对多数场景而言，本方案已足够轻量高效。

总之，面对认证流程这类具有明确上下文依赖和状态管理需求的逻辑，优先选择面向对象封装，而非过度追求装饰器的“魔法感”——清晰胜于炫技，稳健优于巧妙。