You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何延长DefaultAzureCredential获取的Azure令牌有效期或实现令牌自动续订?

如何延长DefaultAzureCredential获取的Azure令牌有效期或实现令牌自动续订?

这个问题我之前做Spark任务对接Azure的时候也碰到过,先给你明确说:没法手动延长Azure AD颁发的令牌有效期——这是Azure AD的安全设计,默认TTL就是1-2小时,而且完全没法通过get_token的参数来修改。不过解决办法很简单,咱们可以实现令牌的自动续订,根本不用你手动盯着过期时间。

下面给你两种实用的解决方案:

1. 优先用Azure官方SDK的客户端类(最省心)

如果你是调用Azure的服务(比如Databricks管理API),尽量用对应的Azure SDK客户端(比如azure-mgmt-databricks里的DatabricksManagementClient)。这些客户端内部已经封装了令牌自动刷新逻辑,你只需要把DefaultAzureCredential传给客户端的构造函数,之后调用API的时候它会自动处理令牌过期的问题,根本不用自己手动调用get_token。

举个简单的例子:

from azure.identity import DefaultAzureCredential
from azure.mgmt.databricks import DatabricksManagementClient

# 初始化凭证和客户端
credential = DefaultAzureCredential()
dbx_client = DatabricksManagementClient(credential, "你的订阅ID")

# 之后直接调用客户端方法就行,令牌刷新全自动
for workspace in dbx_client.workspaces.list():
    print(f"Workspace名称:{workspace.name}")

2. 自己实现令牌自动刷新逻辑(必须手动拿令牌时用)

如果因为某些原因必须手动获取令牌(比如直接调用Databricks的REST API),那可以写一个简单的令牌管理类,用来缓存令牌并在快过期时自动刷新。

示例代码如下:

from azure.identity import DefaultAzureCredential
import time

class TokenManager:
    def __init__(self, scope):
        self.scope = scope
        self.credential = DefaultAzureCredential()
        self.current_token = None
        self.expire_timestamp = 0  # 令牌过期的时间戳

    def get_valid_token(self):
        current_time = time.time()
        # 提前5分钟刷新令牌,避免刚好在请求时过期
        if self.current_token is None or current_time >= self.expire_timestamp - 300:
            # 重新获取令牌
            token_result = self.credential.get_token(self.scope)
            self.current_token = token_result.token
            # 计算过期时间:当前时间 + 令牌有效期(秒)
            self.expire_timestamp = current_time + token_result.expires_on
        return self.current_token

# 使用示例
dbx_scope = "2ff814a6-3304-4ab8-85cb-cd0e6f879c1d/.default"
token_manager = TokenManager(dbx_scope)

# 在Spark任务中,每次需要令牌时调用这个方法就行
import requests

def call_dbx_rest_api(endpoint):
    valid_token = token_manager.get_valid_token()
    headers = {"Authorization": f"Bearer {valid_token}"}
    response = requests.get(endpoint, headers=headers)
    response.raise_for_status()
    return response.json()

这个类会自动检查令牌状态,每次返回的都是有效的令牌,不用担心过期问题。如果是Spark分布式任务,每个executor都可以初始化自己的TokenManager——因为服务主体的环境变量在每个节点上都能访问到。

关于get_token的kwargs参数

你提到的get_token的kwargs,其实是用来传递额外的请求参数给底层的OAuth2.0请求的,比如添加自定义HTTP头、或者传递特定的claims(多租户场景用),但这些参数完全不能改变令牌的TTL。Azure AD的令牌有效期是由租户的安全策略决定的,普通用户没法通过API请求修改,所以不用在这上面花功夫。

备注:内容来源于stack exchange,提问作者BeGreen

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.04.23 13:13:09