Google API:使用 oauth2client.client 从刷新令牌获取凭据

我正在使用谷歌官方oauth2client.client访问谷歌加上 API。我有一个存储在数据库中的刷新令牌（不会过期），并且需要以从中重新创建临时"凭据"（访问令牌）。

但是我找不到一种方法来做到这一点谷歌提供的官方图书馆.

所以我破解了它：使用 urllib 访问给我一个新的 APIaccess_token来自refresh_token。使用access_token然后我可以使用库。

我一定错过了什么！

from apiclient import discovery
from oauth2client.client import AccessTokenCredentials
from urllib import urlencode
from urllib2 import Request , urlopen, HTTPError
import json
# ==========================================
def access_token_from_refresh_token(client_id, client_secret, refresh_token):
  request = Request('https://accounts.google.com/o/oauth2/token',
    data=urlencode({
      'grant_type':    'refresh_token',
      'client_id':     client_id,
      'client_secret': client_secret,
      'refresh_token': refresh_token
    }),
    headers={
      'Content-Type': 'application/x-www-form-urlencoded',
      'Accept': 'application/json'
    }
  )
  response = json.load(urlopen(request))
  return response['access_token']
# ==========================================
access_token = access_token_from_refresh_token(CLIENT_ID, CLIENT_SECRET, REFRESH_TOKEN)
# now I can use the library properly
credentials = AccessTokenCredentials(access_token, "MyAgent/1.0", None)
http = credentials.authorize(httplib2.Http())
service = discovery.build('plus', 'v1', http=http)
google_request = service.people().get(userId='me')
result = google_request.execute(http=http)

我使用：oauth2client.client.GoogleCredentials。

    cred = oauth2client.client.GoogleCredentials(access_token,client_id,client_secret,
                                          refresh_token,expires_at,"https://accounts.google.com/o/oauth2/token",some_user_agent)
    http = cred.authorize(httplib2.Http())
    cred.refresh(http)
    self.gmail_service = discovery.build('gmail', 'v1', credentials=cred)

您可以像这样直接构造一个OAuth2Credentials实例：

import httplib2
from oauth2client import GOOGLE_REVOKE_URI, GOOGLE_TOKEN_URI, client
CLIENT_ID = '<client_id>'
CLIENT_SECRET = '<client_secret>'
REFRESH_TOKEN = '<refresh_token>'
credentials = client.OAuth2Credentials(
    access_token=None,  # set access_token to None since we use a refresh token
    client_id=CLIENT_ID,
    client_secret=CLIENT_SECRET,
    refresh_token=REFRESH_TOKEN,
    token_expiry=None,
    token_uri=GOOGLE_TOKEN_URI,
    user_agent=None,
    revoke_uri=GOOGLE_REVOKE_URI)
credentials.refresh(httplib2.Http())  # refresh the access token (optional)
print(credentials.to_json())
http = credentials.authorize(httplib2.Http())  # apply the credentials

我很容易解决这个问题（你肯定会错过这个文档）。这是我的代码片段，它尝试使用 Picasa API 从活跃用户那里获取所有专辑：

    http = httplib2.Http(ca_certs=os.environ['REQUESTS_CA_BUNDLE'])
    try:
        http = self.oauth.credentials.authorize(http)
        response, album_list = http.request(Picasa.PHOTOS_URL, 'GET')
        if response['status'] == '403':
            self.oauth.credentials.refresh(http)
            response, album_list = http.request(Picasa.PHOTOS_URL, 'GET')
        album_list = json.load(StringIO(album_list))
    except Exception as ex:
        Logger.debug('Picasa: error %s' % ex)
        return {}

使用来自 oauth2client.client.OAuth2Credentials的 refresh 方法。我认为甚至可以使用if response['status'] != '200'.必须检查一下！

如果有人正在寻找如何将刷新令牌与google_auth_oauthlib一起使用的答案，以下内容对我有用：

flow.oauth2session.refresh_token(flow.client_config['token_uri'],
                                 refresh_token=refresh_token,
                                 client_id=<MY_CLIENT_ID>,
                                 client_secret=flow.client_config['client_secret'])
creds = google_auth_oauthlib.helpers.credentials_from_session(
    flow.oauth2session, flow.client_config)

不过，我找不到任何记录这一点的地方。

您也可以

使用 requests 库：

import google.auth.transport.requests
import requests
request = google.auth.transport.requests.Request()
credentials.refresh(request)

这是我在一个活动项目上的示例代码：

acct_creds = {
  'token': self.attachment.account.google_drive_access_token,
  'refresh_token': self.attachment.account.google_drive_refresh_token,
  'client_id': settings.GOOGLE_CLIENT_ID,
  'client_secret': settings.GOOGLE_CLIENT_SECRET,
  'token_uri': 'https://37947.ngrok.io/authenticate/google/callback/',
  'scopes': 'https://www.googleapis.com/auth/drive.appdata https://www.googleapis.com/auth/drive.file https://www.googleapis.com/auth/drive.install',
}
credentials = google.oauth2.credentials.Credentials(**acct_creds)
if credentials.valid:
    print("Credentials valid")
else:
    request = google.auth.transport.requests.Request()
    credentials.refresh(request)

google.auth.transport.requests 模块

我推荐这种方法。

from oauth2client import client, GOOGLE_TOKEN_URI
CLIENT_ID = "client_id"
CLIENT_SECRET = "client_secret"
REFRESH_TOKEN = "refresh_token"

credentials = client.OAuth2Credentials(
    access_token = None, 
    client_id = CLIENT_ID, 
    client_secret = CLIENT_SECRET, 
    refresh_token = REFRESH_TOKEN, 
    token_expiry = None, 
    token_uri = GOOGLE_TOKEN_URI,
    token_ id = None, 
    revoke_uri= None)
http = credentials.authorize(httplib2.Http())

即使访问令牌已过期，由于刷新令牌，凭据仍处于授权状态。

如果你使用 google-auth 的 2018 Youtube Python Quickstart 演示应用，则无法使用 oauth2client 的存储。

所以这是存储凭据的正确方法

以下是 google-auth 的部分工作解决方案，缺少对令牌过期情况的正确处理：

import os
import json
import os.path
import google.oauth2.credentials
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from google_auth_oauthlib.flow import InstalledAppFlow
CLIENT_SECRETS_FILE = "client_secret.json"
SCOPES = ['https://www.googleapis.com/auth/youtube.force-ssl']
API_SERVICE_NAME = 'youtube'
API_VERSION = 'v3'
def get_authenticated_service():
  if os.path.isfile("credentials.json"):
    with open("credentials.json", 'r') as f:
      creds_data = json.load(f)
    creds = Credentials(creds_data['token'])
  else:
    flow = InstalledAppFlow.from_client_secrets_file(CLIENT_SECRETS_FILE, SCOPES)
    creds = flow.run_console()
    creds_data = {
          'token': creds.token,
          'refresh_token': creds.refresh_token,
          'token_uri': creds.token_uri,
          'client_id': creds.client_id,
          'client_secret': creds.client_secret,
          'scopes': creds.scopes
      }
    print(creds_data)
    with open("credentials.json", 'w') as outfile:
      json.dump(creds_data, outfile)
  return build(API_SERVICE_NAME, API_VERSION, credentials = creds)
def channels_list(service, **kwargs):
  results = service.channels().list(**kwargs).execute()
  print('This channel's ID is %s. Its title is %s, and it has %s views.' %
       (results['items'][0]['id'],
        results['items'][0]['snippet']['title'],
        results['items'][0]['statistics']['viewCount']))
   
if __name__ == '__main__':
  os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = '1'
  service = get_authenticated_service()
  channels_list(service, part='snippet,contentDetails,statistics', forUsername='GoogleDevelopers')
  # or if the above doesn't work
  channels_list(service, part='snippet,contentDetails,statistics', id='YOUR_YOUTUBE_CHANNEL_ID')

如果您有刷新令牌，则可以使用 OAuth2Credentials 生成要使用的凭据，如下所示

from googleapiclient.discovery import build
import httplib2
from oauth2client import client, GOOGLE_TOKEN_URI
client_id = "YOUR_CLIENT_ID"
client_secret = "YOUR_CLIENT_SECRET"
refresh_token = "YOUR_REFRESH_TOKEN"
creds = client.OAuth2Credentials(
            access_token = None, 
            client_id = client_id, 
            client_secret = client_secret, 
            refresh_token = refresh_token, 
            token_expiry = None, 
            token_uri = GOOGLE_TOKEN_URI,
            user_agent="pythonclient")
        
creds.refresh(httplib2.Http())

我不知道用户代理中有什么，但我在那里放了一个随机词

现在您可以使用它来构建服务对象并使用谷歌 API

service = build("drive", "v3", credentials=creds)

如果有人想要生成和使用离线刷新令牌以供使用，而不必处理自动化，因为它是为了您的测试，那么使用谷歌 oauth playground 生成一个。查看此视频以获取更多信息。

您可以存储整个凭据，而不仅仅是刷新令牌：

json = credentials.to_json()
credentials = Credentials.new_from_json(json)

查看以这种方式执行此操作的存储对象。

哇..

2 岁的问题，不是一个好的答案..毫不奇怪，因为谷歌文档对此是废话。

执行此操作的正确方法是扩展存储类oauth2client.client.Storage

一个示例实现（使用 mongodb 集合_google_credentials）将是这样的：

class Storage(oauth2client.client.Storage):
def __init__(self, key):
    super(Storage, self).__init__()
    self._key = key
def locked_get(self):
    if not self._key: return None
    data = _google_credentials.find_one({'_id': self._key})
    if not data: return None
    credentials = oauth2client.client.Credentials.new_from_json(json.dumps(data))
    credentials.set_store(self)
    return credentials
def locked_put(self, credentials):
    data = json.loads(credentials.to_json())
    _google_credentials.update_one({'_id': self._key}, {'$set': data}, 
        upsert=True)
    credentials.set_store(self)
def locked_delete(self):
    bucket.delete(self._key)

然后，当您最初在step2_exchange之后获取凭据时，您需要使用以下Storage().put存储它们：

例如：

credentials = flow.step2_exchange(code)
Storage(user_id).put(credentials)

当您再次需要凭据时，只需执行以下操作：

credentials = Storage(user_id).get()

如果您已经有一个Credentials对象，则可以像这样刷新它：

if refresh:
    import google_auth_httplib2
    # credentials instanceof google.oauth2.credentials.Credentials
    credentials.refresh(google_auth_httplib2.Request(httplib2.Http()))

我从一个旧的令牌 JSON 文件创建了Credentials对象，如下所示：

    credentials = google.oauth2.credentials.Credentials(
        token=token_json['access_token'],
        refresh_token=token_json['refresh_token'],
        id_token=token_json['id_token'],
        token_uri=token_json['token_uri'],
        client_id=token_json['client_id'],
        client_secret=token_json['client_secret'],
        scopes=token_json['scopes'])

通过这种方式，我能够改编一些旧的oauth2client代码。

相关内容

最新更新

热门标签：