如何正确刷新Google Cloud Python API凭证？设备流持久化登录异常

API 探路者

2026-6-24

问题

基础背景

开发基于google.cloud Python SDK的GUI应用，对接Google Cloud服务
使用Google OAuth的DeviceClient流实现登录
需求：用户仅登录一次，登录状态本地持久化，重启应用无需重新登录
已掌握设备流流程：获取OAuth响应、保存/加载响应、创建credentials.Credentials及storage.Client
知晓客户端会自动尝试刷新凭证

核心问题

加载本地存储的旧凭证JSON创建Credentials和storage.Client时，客户端报错未授权
重新执行设备流生成的新JSON可正常工作，但写入磁盘重启后再次加载又失效，提示需通过gcloud auth重新登录
尝试传入None作为token、仅提供refresh_token创建Credentials时，调用存储操作立即触发刷新错误
刚获取的凭证调用creds.refresh()也报错，提示需重新认证，即使此时凭证能正常执行存储操作

错误堆栈信息

当传入None作为token调用存储操作时的错误：

File "videoripper/upload.py", line 157, in run
    blob.upload_from_filename(f)
File "venv/lib/python3.12/site-packages/google/cloud/storage/blob.py", line 2959, in upload_from_filename
    self._handle_filename_and_upload(
File "venv/lib/python3.12/site-packages/google/cloud/storage/blob.py", line 2829, in _handle_filename_and_upload
    self._prep_and_do_upload(
File "venv/lib/python3.12/site-packages/google/cloud/storage/blob.py", line 2637, in _prep_and_do_upload
    created_json = self._do_upload(
                   ^^^^^^^^^^^^^^^^
File "venv/lib/python3.12/site-packages/google/cloud/storage/blob.py", line 2443, in _do_upload
    response = self._do_multipart_upload(
               ^^^^^^^^^^^^^^^^^^^^^^^^^^
File "venv/lib/python3.12/site-packages/google/cloud/storage/blob.py", line 1956, in _do_multipart_upload
    response = upload.transmit(
               ^^^^^^^^^^^^^^^^
File "venv/lib/python3.12/site-packages/google/resumable_media/requests/upload.py", line 153, in transmit
    return _request_helpers.wait_and_retry(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "venv/lib/python3.12/site-packages/google/resumable_media/requests/_request_helpers.py", line 155, in wait_and_retry
    response = func()
               ^^^^^^
File "venv/lib/python3.12/site-packages/google/resumable_media/requests/upload.py", line 145, in retriable_request
    result = transport.request(
             ^^^^^^^^^^^^^^^^^^
File "venv/lib/python3.12/site-packages/google/auth/transport/requests.py", line 537, in request
    self.credentials.before_request(auth_request, method, url, request_headers)
File "venv/lib/python3.12/site-packages/google/auth/credentials.py", line 230, in before_request
    self._blocking_refresh(request)
File "venv/lib/python3.12/site-packages/google/auth/credentials.py", line 193, in _blocking_refresh
    self.refresh(request)
File "venv/lib/python3.12/site-packages/google/oauth2/credentials.py", line 431, in refresh
    ) = reauth.refresh_grant(
        ^^^^^^^^^^^^^^^^^^^^^
File "venv/lib/python3.12/site-packages/google/oauth2/reauth.py", line 348, in refresh_grant
    raise exceptions.RefreshError(
google.auth.exceptions.RefreshError: Reauthentication is needed. Please run `gcloud auth application-default login` to reauthenticate.

调用creds.refresh()的错误：

>>> creds.refresh(requests.Request())
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "venv/lib/python3.12/site-packages/google/oauth2/credentials.py", line 431, in refresh
    ) = reauth.refresh_grant(
        ^^^^^^^^^^^^^^^^^^^^^
  File "venv/lib/python3.12/site-packages/google/oauth2/reauth.py", line 348, in refresh_grant
    raise exceptions.RefreshError(
google.auth.exceptions.RefreshError: Reauthentication is needed. Please run `gcloud auth application-default login` to reauthenticate.

现有代码

import json
from oauthlib.oauth2 import DeviceClient
from requests_oauthlib import OAuth2Session
from google.cloud import storage

# Constants
CLIENT_ID = 'xxxxx.apps.googleusercontent.com'
CLIENT_SECRET = 'GOCSPX-xxxxx'
SCOPE = ['https://www.googleapis.com/auth/devstorage.read_write']

# URLs for device and token requests
device_auth_url = 'https://oauth2.googleapis.com/device/code'
token_url = 'https://oauth2.googleapis.com/token'

try:
    with open("/opt/store/creds.json", "w") as f:
        stored = json.load(f)
        creds = credentials.Credentials(
            stored['access_token'],
            refresh_token=stored['refresh_token'],
            token_uri=oa.token_url,
            client_id=oa.CLIENT_ID,
            client_secret=oa.CLIENT_SECRET)
        gclient = storage.Client(project=PROJECT, credentials=creds)
        gclient.list_buckets()
        # it worked
        return stored
except Exception as e:
    # it didn't work
    pass
client = DeviceClient(client_id=CLIENT_ID)
oauth = OAuth2Session(client=client)
device_auth_response = oauth.post(device_auth_url, data={
    'client_id': CLIENT_ID,
    'scope': ' '.join(SCOPE)
})
darj = device_auth_response.json()
device_code = darj['device_code']
user_code = darj['user_code']
verification_url = darj['verification_url']

print(f"Please visit {verification_url} and paste the code: {user_code}", flush=True)

# now wait for the user to go through the flow, polling for success
token_response = oauth.post(token_url, data={
    'client_id': CLIENT_ID,
    'client_secret': CLIENT_SECRET,
    'device_code': device_code,
    'grant_type': 'urn:ietf:params:oauth:grant-type:device_code'
})
if token_response.status_code == 200:
    # Successfully retrieved the token
    token = token_response.json()
    with open("/opt/store/creds.json", "w") as f:
        json.dump(token, f)
    return token
raise Exception("Computer says no.")

解决方案

关键问题分析

凭证加载代码错误：现有代码中加载凭证时用了open("/opt/store/creds.json", "w")（写入模式），导致无法读取文件，每次启动都会直接进入重新登录流程
凭证刷新机制问题：设备流返回的refresh_token需要正确配置token_uri，且刷新时需使用Google官方的凭证刷新逻辑，而非手动调用refresh()时传入错误的请求对象
状态持久化缺失：未监听凭证刷新事件，无法在客户端自动刷新后更新本地存储的凭证

修复步骤

1. 修复凭证加载的文件模式

将读取凭证的代码从写入模式(w)改为读取模式(r)，同时添加文件存在检查。

2. 使用Google官方的凭证存储工具

使用google.oauth2.credentials.Credentials.from_authorized_user_info方法加载凭证，该方法会自动处理刷新逻辑，无需手动构造Credentials对象。

3. 添加凭证刷新监听

在凭证自动刷新后，将新的access_token更新到本地存储（设备流的refresh_token通常不会变化，无需更新）。

4. 正确处理刷新请求

调用creds.refresh()时，需传入google.auth.transport.requests.Request()实例，而非requests.Request()。

修复后的代码

import json
import os
import time
from oauthlib.oauth2 import DeviceClient
from requests_oauthlib import OAuth2Session
from google.cloud import storage
from google.oauth2.credentials import Credentials
from google.auth.transport.requests import Request

# Constants
CLIENT_ID = 'xxxxx.apps.googleusercontent.com'
CLIENT_SECRET = 'GOCSPX-xxxxx'
SCOPE = ['https://www.googleapis.com/auth/devstorage.read_write']
PROJECT = 'your-project-id'
CREDS_PATH = "/opt/store/creds.json"

# URLs for device and token requests
device_auth_url = 'https://oauth2.googleapis.com/device/code'
token_url = 'https://oauth2.googleapis.com/token'

def load_or_refresh_credentials():
    creds = None
    # 检查凭证文件是否存在
    if os.path.exists(CREDS_PATH):
        try:
            with open(CREDS_PATH, 'r') as f:
                stored_info = json.load(f)
                creds = Credentials.from_authorized_user_info(
                    stored_info,
                    scopes=SCOPE
                )
                # 如果凭证过期且有刷新令牌，尝试刷新
                if creds.expired and creds.refresh_token:
                    creds.refresh(Request())
                    # 刷新后更新本地存储的access_token和过期时间
                    stored_info['access_token'] = creds.token
                    stored_info['expiry'] = creds.expiry.isoformat()
                    with open(CREDS_PATH, 'w') as f:
                        json.dump(stored_info, f)
        except Exception as e:
            print(f"加载凭证失败: {str(e)}")
            creds = None

    # 无有效凭证时执行设备流登录
    if not creds or not creds.valid:
        client = DeviceClient(client_id=CLIENT_ID)
        oauth = OAuth2Session(client=client, scope=SCOPE)
        device_auth_response = oauth.post(device_auth_url, data={
            'client_id': CLIENT_ID,
            'scope': ' '.join(SCOPE)
        })
        device_auth_response.raise_for_status()
        darj = device_auth_response.json()
        device_code = darj['device_code']
        user_code = darj['user_code']
        verification_url = darj['verification_url']

        print(f"请访问 {verification_url} 并输入代码: {user_code}", flush=True)

        # 轮询等待用户完成认证
        while True:
            token_response = oauth.post(token_url, data={
                'client_id': CLIENT_ID,
                'client_secret': CLIENT_SECRET,
                'device_code': device_code,
                'grant_type': 'urn:ietf:params:oauth:grant-type:device_code'
            })
            if token_response.status_code == 200:
                token_info = token_response.json()
                # 添加令牌过期时间，方便后续检查
                token_info['expiry'] = (time.time() + token_info['expires_in']).__str__()
                with open(CREDS_PATH, 'w') as f:
                    json.dump(token_info, f)
                creds = Credentials.from_authorized_user_info(token_info, scopes=SCOPE)
                break
            elif token_response.json().get('error') == 'authorization_pending':
                time.sleep(darj['interval'])
            else:
                token_response.raise_for_status()
    
    return creds

# 初始化客户端
creds = load_or_refresh_credentials()
gclient = storage.Client(project=PROJECT, credentials=creds)

# 测试连接
try:
    buckets = list(gclient.list_buckets())
    print(f"成功连接，已获取 {len(buckets)} 个存储桶")
except Exception as e:
    print(f"连接失败: {str(e)}")

额外说明

设备流的refresh_token默认长期有效，除非用户撤销权限或令牌被Google吊销
无需手动检测storage.Client的刷新行为，Credentials对象会自动在请求前检查并刷新令牌
刷新令牌时必须使用google.auth.transport.requests.Request()，该对象包含Google OAuth所需的认证上下文

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

如何正确刷新Google Cloud Python API凭证？设备流持久化登录异常

问题 #

基础背景 #

核心问题 #

错误堆栈信息 #

现有代码 #

解决方案 #

关键问题分析 #

修复步骤 #