如何用Python Requests实现OAuth2 client_credentials自动令牌获取

你可以基于requests.Session封装自定义客户端，结合现成库或手动实现令牌管理逻辑，自动完成令牌获取、过期刷新和401重试。下面提供两种简洁的实现方案：

方案一：基于requests-oauthlib快速实现 #

requests-oauthlib内置了对Client Credentials流的原生支持，无需手动处理令牌请求细节：

首先安装依赖：

pip install requests-oauthlib

封装客户端类：

from requests_oauthlib import OAuth2Session
from oauthlib.oauth2 import BackendApplicationClient

class OAuth2CCClient(OAuth2Session):
    def __init__(self, client_id, client_secret, token_url, **kwargs):
        client = BackendApplicationClient(client_id=client_id)
        super().__init__(client=client, **kwargs)
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = token_url
        # 初始化时自动获取首个令牌
        self.fetch_token()

    def fetch_token(self):
        # 使用Client Credentials模式获取令牌
        token = super().fetch_token(
            token_url=self.token_url,
            client_id=self.client_id,
            client_secret=self.client_secret,
            include_client_id=True
        )
        return token

    def request(self, method, url, **kwargs):
        # 重写请求方法，处理401令牌失效场景
        response = super().request(method, url, **kwargs)
        if response.status_code == 401:
            # 刷新令牌后重试一次请求
            self.fetch_token()
            response = super().request(method, url, **kwargs)
        return response

使用示例：

# 初始化客户端
client = OAuth2CCClient(
    client_id="你的客户端ID",
    client_secret="你的客户端密钥",
    token_url="https://你的认证服务器/token端点"
)

# 像普通Requests一样调用受保护API
response = client.get("https://受保护API地址/资源路径")
print(response.json())

该类继承自OAuth2Session，已自动处理令牌存储、过期判断（依据token的expires_at字段），重写的request方法会在收到401时自动刷新令牌并重试。

方案二：纯手动封装（无额外依赖） #

如果不想引入第三方库，可以基于requests.Session完全手动实现令牌管理逻辑：

import requests
from datetime import datetime, timedelta

class OAuth2ManualClient(requests.Session):
    def __init__(self, client_id, client_secret, token_url):
        super().__init__()
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = token_url
        self.access_token = None
        self.token_expires_at = None

    def _get_new_token(self):
        # 向令牌端点发送POST请求获取新令牌
        data = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }
        response = self.post(self.token_url, data=data)
        response.raise_for_status()  # 令牌请求失败时抛出异常
        token_data = response.json()
        self.access_token = token_data["access_token"]
        # 计算过期时间（预留10秒缓冲，避免网络延迟导致令牌刚好用完）
        expires_in = token_data.get("expires_in", 3600)
        self.token_expires_at = datetime.now() + timedelta(seconds=expires_in - 10)
        # 更新默认请求头
        self.headers.update({"Authorization": f"Bearer {self.access_token}"})

    def _is_token_expired(self):
        return self.access_token is None or datetime.now() >= self.token_expires_at

    def request(self, method, url, **kwargs):
        # 请求前检查令牌有效性，过期则刷新
        if self._is_token_expired():
            self._get_new_token()
        
        # 发送初始请求
        response = super().request(method, url, **kwargs)
        
        # 处理401错误，刷新令牌后重试一次
        if response.status_code == 401:
            self._get_new_token()
            response = super().request(method, url, **kwargs)
        
        return response

使用示例：

client = OAuth2ManualClient(
    client_id="你的客户端ID",
    client_secret="你的客户端密钥",
    token_url="https://你的认证服务器/token端点"
)

# 直接调用API，无需手动处理令牌逻辑
response = client.post("https://受保护API地址/资源路径", json={"key": "value"})
response.raise_for_status()
print(response.json())

注意事项 #

• 401重试仅执行一次，避免因权限不足等非令牌问题陷入无限循环
• 可根据需求添加令牌缓存逻辑（如写入文件或Redis），避免每次重启服务都重新获取令牌
• 令牌请求失败时的异常可根据业务需求捕获并自定义处理

内容的提问来源于stack exchange，提问作者Mr. Developerdude