Python中Access Token过期后如何用Refresh Token调用OneDrive接口

你的核心问题是批量请求过程中Access Token过期，无需手动处理Refresh Token，MSAL库已提供自动刷新机制，只需优化令牌获取逻辑，确保每次请求都使用有效令牌。以下是完整解决方案：

关键优化点 #

• 封装令牌获取函数，每次请求前自动检查令牌有效性，过期则用Refresh Token静默刷新
• 处理请求中的401 Unauthorized错误，触发令牌刷新并重试请求
• 保留令牌缓存持久化，避免重复登录

修改后的代码 #

import os
import atexit
import requests
import urllib.parse
import json
import msal

TENANT_ID = '111*********f5d4'
CLIENT_ID = '7c0*************5517'

AUTHORITY = 'https://login.microsoftonline.com/' + TENANT_ID
base_url = 'https://graph.microsoft.com/v1.0/'
SCOPES = [
    'Files.ReadWrite.All',
    'Sites.ReadWrite.All',
    'User.Read',
    'User.ReadBasic.All'
]

# 初始化令牌缓存
cache = msal.SerializableTokenCache()
if os.path.exists('token_cache.bin'):
    cache.deserialize(open('token_cache.bin', 'r').read())
atexit.register(lambda: open('token_cache.bin', 'w').write(cache.serialize()) if cache.has_state_changed else None)

app = msal.PublicClientApplication(CLIENT_ID, authority=AUTHORITY, token_cache=cache)

def get_valid_access_token():
    """获取有效的Access Token，过期则自动刷新"""
    accounts = app.get_accounts()
    if not accounts:
        # 无缓存账户，触发设备登录
        flow = app.initiate_device_flow(scopes=SCOPES)
        if 'user_code' not in flow:
            raise Exception('Failed to create device flow')
        print(flow['message'])
        result = app.acquire_token_by_device_flow(flow)
    else:
        # 尝试静默刷新令牌（自动用Refresh Token）
        result = app.acquire_token_silent(SCOPES, account=accounts[0])
        if not result:
            # 静默刷新失败，重新登录
            flow = app.initiate_device_flow(scopes=SCOPES)
            print(flow['message'])
            result = app.acquire_token_by_device_flow(flow)
    
    if 'access_token' not in result:
        raise Exception('Failed to get access token')
    return result['access_token']

def make_graph_request(url, token):
    """发送Graph请求，处理401错误并重试"""
    headers = {'Authorization': f'Bearer {token}'}
    response = requests.get(url, headers=headers)
    if response.status_code == 401:
        # 令牌过期，重新获取令牌并重试
        new_token = get_valid_access_token()
        headers['Authorization'] = f'Bearer {new_token}'
        response = requests.get(url, headers=headers)
    response.raise_for_status()  # 抛出其他HTTP错误
    return response.json()

if __name__ == '__main__':
    access_token = get_valid_access_token()
    parent = base_url + '/drives/!FJ9b********************BnEOcC-mq/root:/Box_Migration/Product Images - SHARE'
    link = parent + ":/children"
    
    while True:
        # 获取当前页文件夹列表
        folder_data = make_graph_request(link, access_token)
        for item in folder_data['value']:
            folder_name = item['name']
            print(f"Processing folder: {folder_name}")
            # 获取文件夹内文件
            file_endpoint = base_url + f'/drives/b!FJ9b********************BnEOcC-mq/root:/Box_Migration/Product Images - SHARE/{folder_name}:/children'
            file_data = make_graph_request(file_endpoint, access_token)
            
            # 写入文件
            with open('new_Refresh_token.json', 'a+', encoding='utf-8') as fout:
                json.dump(file_data, fout, indent=4, default=str)
                fout.write(',\n')
        
        # 处理分页
        if "@odata.nextLink" in folder_data:
            link = folder_data["@odata.nextLink"]
        else:
            break

代码说明 #

1. get_valid_access_token函数：

   • 优先从缓存读取账户，尝试静默刷新令牌（MSAL自动调用Refresh Token）
   • 静默刷新失败或无缓存时，触发设备登录流程
   • 确保返回的令牌始终处于有效状态

2. make_graph_request函数：

   • 封装请求逻辑，自动捕获401错误（令牌过期）
   • 遇到401时重新获取令牌并重试，避免批量任务中断

3. 缓存持久化：

   • 令牌缓存自动保存到token_cache.bin，下次启动无需重新登录
   • atexit确保程序退出时保存最新缓存状态

注意事项 #

• 确保token_cache.bin文件有读写权限，防止缓存丢失
• 若遇权限错误，检查Azure应用的API权限是否已获得管理员同意
• 批量请求可添加1-2秒延迟，避免触发Graph API速率限制

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