7

我花了几个小时试图解决如何使用 python 请求库获取 oauth2 令牌。我究竟做错了什么？

  r=requests.post("https://accounts.google.com/o/oauth2/token",
                            headers={
                                'content-type':'application/x-www-form-urlencoded'},
                            params={
                                'code':request.GET['code'],
                    'client_id':'11111111111.apps.googleusercontent.com',
                                'client_secret':'my secert',
                                'redirect_uri':'http://localhost/etc'
                                'grant_type':'authorization_code'})
4

3 回答 3

3

请求的数据必须在正文中发送，而不是在查询字符串中

在您的代码中更改params=为。data=

有关详细信息，请参阅此讨论： https ://groups.google.com/forum/#!topic/oauth2-dev/ahP0cz2ao0o

于 2013-08-02T10:16:56.917 回答
3

这是一个非常晚的答案，但可能对某些人有用。

我已经让 Google OAuth2 流程与这个简单类中的请求很好地配合

from requests.adapters import HTTPAdapter
from requests_oauthlib import OAuth2Session
from urllib3.util.retry import Retry
import os

from yaml import load, dump, YAMLError

try:
    from yaml import CLoader as Loader, CDumper as Dumper
except ImportError:
    from yaml import Loader, Dumper


# OAuth endpoints given in the Google API documentation
authorization_base_url = "https://accounts.google.com/o/oauth2/v2/auth"
token_uri = "https://www.googleapis.com/oauth2/v4/token"

'''
Defines a very simple class to handle google api authorization flow
for the requests library. Includes saving the token and automatic
token refresh.
giles 2018
'''


class Authorize:
    def __init__(self, scope, token_file, secrets_file):
        self.scope = scope
        self.token_file = token_file
        self.session = None
        self.token = None
        try:
            with open(secrets_file, 'r') as stream:
                all_yaml = load(stream, Loader=Loader)
            secrets = all_yaml['installed']
            self.client_id = secrets['client_id']
            self.client_secret = secrets['client_secret']
            self.redirect_uri = secrets['redirect_uris'][0]
            self.token_uri = secrets['token_uri']
            self.extra = {
                'client_id': self.client_id,
                'client_secret': self.client_secret}

        except (YAMLError, IOError):
            print('missing or bad secrets file: {}'.format(secrets_file))
            exit(1)

    def load_token(self):
        try:
            with open(self.token_file, 'r') as stream:
                token = load(stream, Loader=Loader)
        except (YAMLError, IOError):
            return None
        return token

    def save_token(self, token):
        with open(self.token_file, 'w') as stream:
            dump(token, stream, Dumper=Dumper)
        os.chmod(self.token_file, 0o600)

    def authorize(self):
        token = self.load_token()

        if token:
            # force refresh on load token.expires_in = -30
            #  todo this is no longer in the token ??
            #   how to force update?
            self.session = OAuth2Session(self.client_id, token=token,
                                         auto_refresh_url=self.token_uri,
                                         auto_refresh_kwargs=self.extra,
                                         token_updater=self.save_token)
        else:
            print(self.scope)
            self.session = OAuth2Session(self.client_id, scope=self.scope,
                                         redirect_uri=self.redirect_uri,
                                         auto_refresh_url=self.token_uri,
                                         auto_refresh_kwargs=self.extra,
                                         token_updater=self.save_token)

            # Redirect user to Google for authorization
            authorization_url, _ = self.session.authorization_url(
                authorization_base_url,
                access_type="offline",
                prompt="select_account")
            print('Please go here and authorize,', authorization_url)

            # Get the authorization verifier code from the callback url
            response_code = input('Paste the response token here:')

            # Fetch the access token
            self.token = self.session.fetch_token(
                self.token_uri, client_secret=self.client_secret,
                code=response_code)
            self.save_token(self.token)

        # note this does retries on POST as well
        # CAUTION - this might not be desirable for all applications
        retries = Retry(total=5,
                        backoff_factor=0.1,
                        status_forcelist=[500, 502, 503, 504],
                        method_whitelist=frozenset(['GET', 'POST']),
                        raise_on_status=False)
        self.session.mount('https://', HTTPAdapter(max_retries=retries))
于 2019-01-26T22:48:07.293 回答
0

我在学习用于 google 计算请求的 python API 时看到了这篇文章，发现它准确但陈旧；现在大部分代码已经集成到谷歌库中。获取/使用令牌的一种方法：

auth_req = google.auth.transport.requests.Request()
creds.refresh(auth_req)
print(f'token: {creds.token} : {creds.valid} {creds.expiry}')

# read all disks under project/zone
url='https://compute.googleapis.com/compute/v1/projects/prog-name/zones/zone-name/disks'
headers_api = {'Authorization': 'Bearer ' + creds.token}
resp = requests.get(url=url, headers=headers_api)

for disk in resp.json()['items']:
    print(f"{disk['name']}: {disk['sizeGb']}")

这在google.cloud.compute_v1API 中得到进一步巩固：

project='prog-name'
zone='zone-name'

# get a context for the API client
client = compute_v1.InstancesClient()
# get the VMs in the project/zone 
clist = client.list(project=project, zone=zone)

# print each VM
for vm in clist.pages:
    print("VM: {}".format(vm))
于 2022-02-15T17:45:30.080 回答