如何配置OpenAPI Generator Python客户端实现OAuth2自动令牌获取

核心逻辑 #

由于服务器不返回刷新令牌，我们需要实现一套令牌管理机制：在首次调用、令牌过期、收到401认证失败响应时，自动调用/oauth2/token接口获取新令牌。以下是具体实现步骤：

1. 生成客户端时指定关键参数 #

生成客户端时，明确指定使用pydantic-v1，并启用Basic Auth支持，命令示例如下：

openapi-generator-cli generate \
  -i your-swagger-spec.yaml \
  -g python \
  --additional-properties=pydanticV1=true,packageName=your_client \
  --auth basic

2. 编写令牌管理器类 #

创建一个专门的类来处理令牌的获取、缓存和过期校验：

import time
from typing import Optional
from your_client.api_client import ApiClient
from your_client.configuration import Configuration
from your_client.api.o_auth2_api import OAuth2Api

class TokenManager:
    def __init__(self, config: Configuration):
        self.config = config
        self.access_token: Optional[str] = None
        self.token_expiry_time: float = 0.0

    def _fetch_new_token(self):
        # 初始化API客户端用于请求令牌
        api_client = ApiClient(configuration=self.config)
        oauth_api = OAuth2Api(api_client)
        
        # 发起Client Credentials令牌请求（Basic Auth已通过config的用户名密码自动处理）
        token_resp = oauth_api.oauth2_token_post(grant_type="client_credentials")
        
        self.access_token = token_resp.access_token
        # 计算令牌过期时间（当前时间+返回的有效期）
        self.token_expiry_time = time.time() + token_resp.expires_in

    def get_valid_token(self):
        # 检查令牌是否不存在或已过期，是则重新获取
        if not self.access_token or time.time() >= self.token_expiry_time:
            self._fetch_new_token()
        return self.access_token

3. 扩展ApiClient实现自动令牌注入与重试 #

继承生成的ApiClient，重写请求方法，实现令牌自动注入和401重试逻辑：

from your_client.api_client import ApiClient
import requests

class AuthenticatedApiClient(ApiClient):
    def __init__(self, token_manager: TokenManager, **kwargs):
        super().__init__(**kwargs)
        self.token_manager = token_manager

    def request(self, method, url, query_params=None, headers=None, post_params=None, body=None, _preload_content=True, _request_timeout=None):
        # 注入有效的Bearer令牌
        headers = headers or {}
        headers["Authorization"] = f"Bearer {self.token_manager.get_valid_token()}"

        try:
            return super().request(method, url, query_params, headers, post_params, body, _preload_content, _request_timeout)
        except requests.exceptions.HTTPError as e:
            # 捕获401错误，刷新令牌后重试一次
            if e.response.status_code == 401:
                self.token_manager._fetch_new_token()
                headers["Authorization"] = f"Bearer {self.token_manager.get_valid_token()}"
                return super().request(method, url, query_params, headers, post_params, body, _preload_content, _request_timeout)
            # 其他HTTP错误直接抛出
            raise

4. 初始化并使用客户端 #

from your_client.configuration import Configuration

# 初始化配置，设置Basic Auth的用户名和密码
config = Configuration()
config.username = "your_username"
config.password = "your_password"

# 初始化令牌管理器和自定义认证客户端
token_manager = TokenManager(config)
authenticated_client = AuthenticatedApiClient(token_manager, configuration=config)

# 使用自定义客户端初始化业务API类
from your_client.api.your_business_api import YourBusinessApi
business_api = YourBusinessApi(api_client=authenticated_client)

# 调用API时会自动处理令牌的获取、刷新逻辑
response = business_api.your_endpoint_get()

注意事项 #

• 确保你的OpenAPI规范中正确定义了/oauth2/token端点和OAuth2安全方案，生成的OAuth2Api类能正常调用令牌接口。
• 如果令牌响应没有expires_in字段，需要根据服务器约定的令牌有效期手动设置过期时间。
• 重试逻辑仅做一次，避免无限重试导致的资源浪费。

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