使用服务帐户模拟 Google Drive 获取刷新令牌

问题描述

我编写了一个脚本,用于在我们公司的多个用户 gdrives 中搜索文件。问题是该脚本可以正常工作,但只能运行一段时间,然后它会出现 HTTP 401 错误,这很可能是由于访问令牌过期。我正在使用启用了域范围委托的服务帐户,并使用 google.oauth2 python 库来创建 Drive 服务对象(请参阅下面的代码)。我想以编程方式获取刷新令牌并在当前令牌过期时生成新的访问令牌。问题是没有关于如何使用服务帐户做到这一点的文档。有正常的用户交互式 oauth 流程。我的问题是我将如何使用驱动器对象来检索刷新令牌,然后创建新的访问令牌。我知道我将不得不以某种方式使用我的服务帐户的私钥,但不确定。

我一直在阅读这两个文档。 https://developers.google.com/identity/protocols/oauth2 https://developers.google.com/identity/protocols/oauth2/service-account

  • 使用模拟用户创建驱动器服务对象
from google.oauth2 import service_account
import googleapiclient.discovery

def impersonate_user(user_to_impersonate,key_file,domain):
        scopes = ['https://www.googleapis.com/auth/drive',]

        if user_to_impersonate is None:
            raise InvalidDomainUser(f"{user_to_impersonate} is not a member of {domain}")

        if key_file is None:
            raise FileNotFoundError(f"{key_file} is not a valid service account file")

        delegated_credentials = service_account.Credentials.from_service_account_file(
            key_file,scopes=scopes)
        # Impersonate User.
        delegated_credentials = delegated_credentials.with_subject(user_to_impersonate)
        drive_service = googleapiclient.discovery.build('drive','v2',credentials=delegated_credentials)

        # Set new drive resource object
        return drive_service
service = impersonate_user(user_to_impersonate,domain):
children = service.children().list(folderId=folderId,maxResults=1000,**param).execute() 
for child in items:
            item = service.files().get(fileId=child['id']).execute()
            print(item.get("title",None))


解决方法

模拟用户会生成一个过期的访问令牌。要获得新的访问令牌,您只需按照第一次的方式生成一个。服务帐户没有刷新令牌的概念,您只需使用您的凭据生成一个新的访问令牌。

来源:内置 Google Service Account support at Xkit

,

这是带有python的服务帐户的基本代码。您需要添加委托部分。

只要您使用相同的服务,图书馆就会在需要时为您获取新的访问令牌。它应该不会过期。

"""Hello drive"""

from apiclient.discovery import build
from oauth2client.service_account import ServiceAccountCredentials


SCOPES = ['https://www.googleapis.com/auth/drive.readonly']
KEY_FILE_LOCATION = '<REPLACE_WITH_JSON_FILE>'
VIEW_ID = '<REPLACE_WITH_VIEW_ID>'


def drive():
  """Initializes an drive service object.

  Returns:
    An authorized drive service object.
  """
  credentials = ServiceAccountCredentials.from_json_keyfile_name(
      KEY_FILE_LOCATION,SCOPES)

  # Build the service object.
  drive = build('drive','v3',credentials=credentials)

  return drive