如何正确刷新 Google Cloud Python API 凭据？

Question

我有一个用 Python 和 tkinter 编写的桌面应用程序。 这个桌面应用程序可以做很多事情，包括将文件上传到谷歌存储桶。 该应用程序应该有很多天的正常运行时间，并且理想情况下允许重新启动，而无需在经过身份验证后重新登录。

我有一个运行良好的“设备”oauth 登录流程——它给了我一个刷新令牌和一个身份验证令牌。 我以相当安全的方式在本地保存这些参数，因为刷新令牌是合法的凭据。 （尽管如果它们被偷了，至少下次刷新会发现这一点，并且也许能够对此采取措施。）

身份验证令牌可以工作一个小时。然后就需要刷新了。 这是文档对我没有帮助的地方。如何正确刷新我拥有的凭据，同时：

1. 谷歌存储客户端会看到更新后的身份验证令牌以供将来的请求。
2. 调用刷新的代码会看到新交换的刷新令牌，这样我就可以将其保存在本地，这样下次应用程序启动仍然可以工作。

我可以让谷歌存储客户端自动刷新凭据。但是，当它这样做时，我找不到将刷新令牌返回以进行持久存储的方法，因此应用程序的下一次启动仍然可以工作。 我也找不到一个好的方法来知道刷新何时发生，以便我可以知道适当地存储令牌。

必填代码：

from google.cloud import storage
from google.oauth2 import credentials

import oa

...

def make_client_from_token(token):
    creds = credentials.Credentials(token['access_token'],
                                    refresh_token=token['refresh_token'],
                                    token_uri=oa.token_url,
                                    client_id=oa.CLIENT_ID,
                                    client_secret=oa.CLIENT_SECRET)
    return storage.Client(credentials=creds, project=PROJECT)

我假设刷新令牌在使用时失效并替换为新令牌，这似乎与我看到的行为相符，但如果这是一个错误的假设，我也想知道！

那么，我错过了什么？

Answer 1

我很好奇为什么你要对刷新令牌进行硬编码。当应用程序运行时，您应该提示用户授权您的应用程序，然后存储访问令牌和刷新令牌。

假设您有一个有效的刷新令牌，库将自动刷新它并为您提供一个永不访问令牌。但是，如果您的刷新令牌无效，那么它应该提示用户再次授权应用程序。

这可以使用以下代码来完成。

def main():
    """Shows basic usage of the Google API.
    """
    creds = None
    # The file token.json stores the user's access and refresh tokens, and is
    # created automatically when the authorization flow completes for the first
    # time.
    if os.path.exists('token.json'):
        creds = Credentials.from_authorized_user_file('token.json', SCOPES)
    # If there are no (valid) credentials available, let the user log in.
    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
        else:
            flow = InstalledAppFlow.from_client_secrets_file(
                credentials_path, SCOPES)
            creds = flow.run_local_server(port=0)
        # Save the credentials for the next run
        with open('token.json', 'w') as token:
            token.write(creds.to_json())

    try:
        client = storage.Client(credentials=creds, project=PROJECT)

        print(client.bucket('daimto-tutorials'))


    except HttpError as error:
        print('An error occurred: %s' % error)