Aiogoogle

Async Google API Client

Aiogoogle makes it possible to access most of Google’s public APIs which include:

Google Calendar API
Google Drive API
Google Contacts API
Gmail API
Google Maps API
Youtube API
Translate API
Google Sheets API
Google Docs API
Gogle Analytics API
Google Books API
Google Fitness API
Google Genomics API
Google Cloud Storage
Kubernetes Engine API
And more.

Library Setup ⚙️

$ pip install aiogoogle

Google Account Setup

Create a project: Google’s APIs and Services dashboard.
Enable an API: API Library.
Create credentials: Credentials wizard.
Pick an API: Google’s APIs Explorer

Authentication

Google APIs can be called on behalf of 3 main principals:

User account
Service account
Anonymous principal by using: API keys

User account

Should be used whenever the application wants to access information tied to a Google user. Google provides two main authorization/authentication strategies that will enable your application act on behalf of a user account:

OAuth2
OpenID Connect

OAuth2

OAuth2 is an authorization framework used to allow a client (an app) to act on behalf of a user. It isn’t designed to identify who the user is, rather only defines what a client (the app) can access.

OAuth2 has 4 main flows. The most popular of them and the only one supported by Google is Authorization Code Flow.

There are 3 main parties involved in this flow:

User:
- represented as UserCreds.
Client:
- represented as ClientCreds.
Resource Server/Authorization server:
- The service that aiogoogle acts as a client to. e.g. Calendar, Youtube, etc.

Quick auth script

Here’s a script that will help you get an access and refresh token for your personal Google account.

Link: https://github.com/omarryhan/aiogoogle/blob/master/examples/auth/oauth2.py

Steps to run:

Clone the repo
cd aiogoogle/examples
cp _keys.yaml keys.yaml
Fillout the following:
- client_id
- client_secret
- scopes
cd aiogoogle/examples/auth
run python oauth2.py
After the webbrowser opens and you authorize your app, a UserCreds JSON object will be returned to your browser.
Copy the access token and refresh token and either paste them in keys.yaml if you’ll be using the examples directory. Or copy and paste them in your own code.

Integrate OAuth2 in an existing web app

If you want to integrate OAuth2 in an existing web app, or use it on many users (not just your personal account), then follow these steps:

Generate an authentication URL and send it to the user you wish to access their data on their behalf.

@app.route('/authorize')
def authorize(request):
    uri = aiogoogle.oauth2.authorization_url(
        client_creds={
            'client_id': '...',
            'client_secret': '...',
            'scopes': [
                '...',
                '...'
            ],
            'redirect_uri': 'http://localhost:5000/callback/aiogoogle'
        },
    )
    return response.redirect(uri)

Now, the user should get redirected to Google’s auth webpage, were they will be prompted to give your app the authorization it requested.
After the user authorizes your app, the user should get redirected back to your domain (to the redirect_uri you specified in step 1) giving you a grant code (not to be confused with an access token). Using this grant code, your application should then request a UserCreds dict which will contain an access and a refresh token.

@app.route('/callback/aiogoogle')
async def callback(request):
    # First, check if there's an error
    if request.args.get('error'):
        error = {
            'error': request.args.get('error'),
            'error_description': request.args.get('error_description')
        }
        return response.json(error)

    # Here we request the access and refresh token
    elif request.args.get('code'):
        full_user_creds = await aiogoogle.oauth2.build_user_creds(
            grant = request.args.get('code'),
            client_creds = CLIENT_CREDS
        )
        # Here, you should store full_user_creds in a db. Especially the refresh token and access token.
        return response.json(full_user_creds)

    else:
        # Should either receive a code or an error
        return response.text("Something's probably wrong with your callback")

Warning

You shouldn’t hand the user of your app their access and refresh tokens.

This is only done here for convenience and for personal use.

Ideally, you’d want to store their tokens in a database on your backend.

OpenID Connect

OpenID Connect is an authentication layer built on top of OAuth2 (an authorization framework). This method should be used when the client (the app) wants to identify the user i.e. authenticate them. Since OpenID connect is a superset of OAuth2, this method will also give the client the authorization needed to edit resources of the user.

In more practical terms, using OAuth2 alone will only return you a token that can be used to access the data with the scope that the app requested. Using OpenIDConnect will return the same access token as with OAuth2 plus an ID token JWT of the user. This ID token JWT will contain “claims” about the user which your app will need to properly know who they are. Here’s an example of how an ID token JWT should look like.

Hint

OpenID Connect should be used if you’re implementing “social signin”.

Quick auth script

Here’s a script that can help you get an access & refresh token + OpenID connect claims for your personal Google account.

Link: https://github.com/omarryhan/aiogoogle/blob/master/examples/auth/openid_connect.py

Steps to run:

Clone the repo
cd aiogoogle/examples
cp _keys.yaml keys.yaml
Fillout the following:
- client_id
- client_secret
- scopes
cd aiogoogle/examples/auth
run python openid_connect.py
After the webbrowser opens and you authorize your app, a UserCreds JSON object and a User Info JSON object will be returned to your browser.
Copy the access token and refresh token and either paste them in keys.yaml if you’ll be using the examples directory. Or copy and paste them in your own code.

Integrate OpenID Connect in an existing web app

This works just like OAuth2 but with a few differences.

As before, we first redirect the user to the authorization prompt page.

@app.route('/authorize')
def authorize(request):
    uri = aiogoogle.openid_connect.authorization_url(
        client_creds={
            'client_id': '...',
            'client_secret': '...',
            'redirect_uri': '...',
            'scopes': [
                '...'
            ]
        },
        nonce='...'  # Random value that prevents replay attacks
    )
    return response.redirect(uri)

After the user authorizes your app and gets redirected back to your domain, you should then request the access and refresh token.

The difference here is that aside from the full_user_creds dict, you also get the full_user_info dict. This dict has claims about the user. Here’s an example full_user_info dict:

{
    "iss": "https://accounts.google.com",
    "azp": "1234987819200.apps.googleusercontent.com",
    "aud": "1234987819200.apps.googleusercontent.com",
    "sub": "10769150350006150715113082367",
    "at_hash": "HK6E_P6Dh8Y93mRNtsDB1Q",
    "hd": "example.com",
    "email": "jsmith@example.com",
    "email_verified": "true",
    "iat": 1353601026,
    "exp": 1353604926,
    "nonce": "0394852-3190485-2490358"
}

You should use the sub claim to uniquely identify your user. Google guarantees that this claim will be unique, unlike the email (common mistake).

If you want to understand what the rest of the claims are used for, please head here.

@app.route('/callback/aiogoogle')
async def callback(request):
    if request.args.get('error'):
        error = {
            'error': request.args.get('error'),
            'error_description': request.args.get('error_description')
        }
        return response.json(error)

    elif request.args.get('code'):
        # Returns an access and refresh token
        full_user_creds = await aiogoogle.openid_connect.build_user_creds(
            grant=request.args.get('code'),
            client_creds=CLIENT_CREDS,
            nonce=nonce,  # Same nonce as above
            verify=False
        )

        # A dict having claims of the user e.g. the sub claim and iat claim.
        # Check https://developers.google.com/identity/protocols/oauth2/openid-connect#an-id-tokens-payload for more info
        full_user_info = await aiogoogle.openid_connect.get_user_info(full_user_creds)

        # Here you should save both full_user_creds and full_user_info to a db if your app will be serving multiple users.
        return response.text(
            f"full_user_creds: {pprint.pformat(full_user_creds)}\n\nfull_user_info: {pprint.pformat(full_user_info)}"
        )

    else:
        # Should either receive a code or an error
        return response.text("Something's probably wrong with your callback")

Warning

You should neither hand the user of your app their tokens nor their OpenID Connect claims.

This is only done here for convenience and for personal use.

Ideally, you’d want to store their info in a database on your backend.

Service account

A service account is a special kind of account that belongs to an application or a virtual machine (VM) instance but not a person. They are intended for scenarios where your application needs to access resources or perform actions on its own, such as running App Engine apps or interacting with Compute Engine instances.

There are a couple of authentication/authorization methods you can use with service accounts. We’ll only concern ourselves with the ones that will grant us access to Google APIs and not for any other purpose e.g. communicating with other servers on a different cloud, as this is out of scope.

OAuth2

OAuth2 is the most commonly used service account authorization method and the only one implemented by Aiogoogle. There are a couple of ways you can access Google APIs using OAuth2 tokens for service accounts:

1. By passing a user managed service account key:

Full example here.

import json
import asyncio
from aiogoogle import Aiogoogle
from aiogoogle.auth.creds import ServiceAccountCreds


service_account_key = json.load(open('test_service_account.json'))

creds = ServiceAccountCreds(
    scopes=[
        "https://www.googleapis.com/auth/devstorage.read_only",
        "https://www.googleapis.com/auth/devstorage.read_write",
        "https://www.googleapis.com/auth/devstorage.full_control",
        "https://www.googleapis.com/auth/cloud-platform.read-only",
        "https://www.googleapis.com/auth/cloud-platform",
    ],
    **service_account_key
)


async def list_storage_buckets(project_id=creds["project_id"]):
    async with Aiogoogle(service_account_creds=creds) as aiogoogle:
        storage = await aiogoogle.discover("storage", "v1")
        res = await aiogoogle.as_service_account(
            storage.buckets.list(project=creds["project_id"])
        )
        print(res)


if __name__ == "__main__":
    asyncio.run(list_storage_buckets())

2. By pointing the GOOGLE_APPLICATION_CREDENTIALS environment variable at the location of the JSON key file.

Full example here.

First, set the environment variable, if not already set:

export GOOGLE_APPLICATION_CREDENTIALS="location_of_key_file.json"

Then:

import asyncio
import print
from aiogoogle import Aiogoogle
from aiogoogle.auth.creds import ServiceAccountCreds

creds = ServiceAccountCreds(
    scopes=[
        "https://www.googleapis.com/auth/devstorage.read_only",
        "https://www.googleapis.com/auth/devstorage.read_write",
        "https://www.googleapis.com/auth/devstorage.full_control",
        "https://www.googleapis.com/auth/cloud-platform.read-only",
        "https://www.googleapis.com/auth/cloud-platform",
    ],
)


async def list_storage_buckets(project_id=creds["project_id"]):
    aiogoogle = Aiogoogle(service_account_creds=creds)

    # Notice this line. Here, Aiogoogle loads the service account key.
    await aiogoogle.service_account_manager.detect_default_creds_source()

    async with aiogoogle:
        storage = await aiogoogle.discover("storage", "v1")
        res = await aiogoogle.as_service_account(
            storage.buckets.list(project=creds["project_id"])
        )

    print(res)

if __name__ == "__main__":
    asyncio.run(list_storage_buckets())

3. Automatic detection from Google Cloud SDK (Not supported yet):

This should call the Google Cloud SDK CLI and ask it for an access token without passing service_account_creds.

4. Automatic detection from Google App Engine environment (Not supported yet):

This should return an OAuth2 token for the service account and cache it for the App Engine application without passing service_account_creds.

5. Automatic detection from Google Compute Engine/Google Cloud Run/Google Cloud Functions environment

This should return an OAuth2 token from the Google Compute Engine metadata server, which means that you don’t have to pass service_account_creds, unless you want to specify scopes.

This should work the same as #2, but skip the part were you set the environment variable.

Full example here.

Others

Other than the OAuth2 method, there are:

OpenID connect ID token (OIDC ID token)
Self-signed JWT credentials
Self-signed On-demand JWT credentials
Short lived tokens.

But they are not supported and you might want to consider another library if you want to use them.

Anonymous principal (API keys)

An API key is a simple string that you can get from Google Cloud Console.

Using an API key is suitable for when you only want to access public data.

Example:

async with Aiogoogle(api_key='...') as aiogoogle:
    google_api = await aiogoogle.discover('google_api', 'v1')
    result = await aiogoogle.as_api_key(
        google_api.foo.bar(baz='foo')
    )

Discovery Service

Most of Google’s public APIs are documented/discoverable by a single API called the Discovery Service.

Google’s Discovery Serivce provides machine readable specifications known as discovery documents (similar to Swagger/OpenAPI). e.g. Google Books.

Aiogoogle is a Pythonic wrapper for discovery documents.

For a list of supported APIs, visit: Google’s APIs Explorer.

Discovery docs and the `Aiogoogle` object explained

Intro

Now that you have figured out which authentication scheme you are going to use and got your hands on an access and a refresh token, let’s make some API calls.

Assuming that you chose the: urlshortener v1 API:

Note

As of March 30, 2019, the Google URL shortening service was shut down. So this only serves as an example.

Create a URL-shortener Google API instance

Let’s discover the urlshortener discovery document and create an Aiogoogle representation of it and call it url_shortener

import asyncio
from aiogoogle import Aiogoogle

async def create_api(name, version):
    async with Aiogoogle(
        user_creds={'access_token': '...', 'refresh_token': '...'}  # Or use the UserCreds object. Same thing.
    ) as aiogoogle:
        # Downloads the API specs and creates an API object
        return await aiogoogle.discover(name, version)

url_shortener = asyncio.run(
    create_api('urlshortener', 'v1')
)

Structure of an API

This is what the JSON representation of the discovery document we downloaded looks like.

Root
|__ name
|__ version
|__ baseUrl
|__ Global Parameters
|__ Schemas
|__ Resources
    |__ Resources
        |_ Resources
            |_...
        |_ Methods
    |__ Methods
|__ Methods
    |_ Path
    |_ Parameters
    |_ Request Body
    |_ Response Body

Full reference:.

You don’t have to worry about most of this. What’s important to understand right now is that a discovery service is just a way to specify the resources and methods of an API.

What are resources and methods?

It’s easier to explain this with an example. In the youtube-v3 API, a resource can be: videos and a method would be: list.

The way you would access the list method is by typing: youtube_v3.videos.list()

Some rules:

A resource can also have nested resources. e.g. youtube_v3.videos.comments.list()

There can be top level methods that are not associated with any resource. However, that’s not common.

Finally, the only way you can get data from a Google API is by calling a method. You can’t call a resource and expect data.

Next, we will programatically browse an API

Browse an API - Programatically

Back to the URL shortener API.

Let’s list the resources of the URL shortener API

>>> url_shortener['resources']

{
    'url': {

        'methods':

            'get': ...
            'insert': ...
            'list': ...
}

Now, let’s browse the resource called url

>>> url_resource = url_shortener.url

>>> url_resource.methods_available

It has the following methods available to call:

['get', 'insert', 'list']

Sometimes resources have nested resources

>>> url_resource.resources_available

[]

This one doesn’t.

Let’s inspect the method called list of the url resource

>>> list_url = url_resource.list

Let’s see what this method does

>>> list_url['description']

"Retrieves a list of URLs shortened by a user."

Cool, now let’s see the optional parameters that this method accepts

>>> list_url.optional_parameters

['projection', 'start_token', 'alt', 'fields', 'key', 'oauth_token', 'prettyPrint', 'quotaUser']

And the required parameters

>>> list_url.required_parameters

[]

Let’s check out what the start_token optional parameter is and how it should look like

>>> list_url.parameters['start_token']

{
    "type": "string",
    "description": "Token for requesting successive pages of results.",
    "location": "query"
}

Finally let’s create a request, that we’ll then send with Aiogoogle

>>> request = list_url(start_token='a_string', key='a_secret_key')

# Equivalent to:

>>> request = url_shortener.url.list(start_token='a_start_token', key='a_secret_key')

Here we passed the url.list method the parameters we want and an unsent request has been created

We can inspect the URL of the request by typing:

>>> request.url

'https://www.googleapis.com/url/history?start_token=a_start_token&key=a_secret_key'

Browse an API - Manually

Sometimes it’s easier to browse a discovery document manually instead of doing it through a Python terminal. Here’s how to do it:

Download a JSON viewer on your browser of choice. e.g:

https://chrome.google.com/webstore/detail/json-viewer/gbmdgpbipfallnflgajpaliibnhdgobh?hl=en-US
Put your API name and version in this link and open the link in your browser:

https://www.googleapis.com/discovery/v1/apis/{api}/{version}/rest

e.g.

https://www.googleapis.com/discovery/v1/apis/youtube/v3/rest
You’ll then get a human readable JSON document with a structure similar to the one you’ve seen above.

Let’s try to get a Youtube video rating using the “youtube-v3” api:

Expand the resources property
Expand the videos property (Which is a resource)
Expand the methods property to see what methods this resource has
Expand the getRating property (a method)
Expand the parameters property to see the arguments that this method accepts
Expand the id property (a parameter) which happens to be the only parameter being accepted
Read the required property to see whether or not this property is required
Read the type property to see the type of this parameter
Now to create an unsent request using this method, you can type: youtube_v3.videos.getRating(id='an_id')

I hope you spot the pattern by now :)

Next we’ll be sending all the unsent requests that we’ve been creating.

Send a Request

Let’s create a coroutine that shortens URLs using an API key

When creating an Aiogoogle object, it defaults to using an Aiohttp session to send your requests with. More on changing the default session in this section.

import asyncio
from aiogoogle import Aiogoogle
from pprint import pprint

api_key = 'you_api_key'

async def shorten_urls(long_url):
    async with Aiogoogle(api_key=api_key) as aiogoogle:
        url_shortener = await aiogoogle.discover('urlshortener', 'v1')
        short_url = await aiogoogle.as_api_key(  # the request is being sent here
            url_shortener.url.insert(
                json=dict(
                    longUrl=long_url
                )
            )
        )

    return short_url

short_url = asyncio.run(shorten_urls('https://www.google.com'))
pprint(short_url)

{
    "kind": "urlshortener#url",
    "id": "https://goo.gl/Dk2j",
    "longUrl": "https://www.google.com/"
}

Send Requests Concurrently

Now let’s shorten two URLs at the same time

import asyncio
from aiogoogle import Aiogoogle
from pprint import pprint

async def shorten_url(long_urls):
    async with Aiogoogle(api_key=api_key) as aiogoogle:
        url_shortener = await aiogoogle.discover('urlshortener', 'v1')
        short_urls = await aiogoogle.as_api_key(

            url_shortener.url.insert(
                json=dict(
                    longUrl=long_url[0]
                ),

            url_shortener.url.insert(
                json=dict(
                    longUrl=long_url[1]
                )
        )
    return short_urls

short_urls = asyncio.run(
    shorten_url(
        ['https://www.google.com', 'https://www.google.org']
    )
)
pprint(short_urls)

[
    {
        "kind": "urlshortener#url",
        "id": "https://goo.gl/Dk2j",
        "longUrl": "https://www.google.com/"
    },
    {
        "kind": "urlshortener#url",
        "id": "https://goo.gl/Dk23",
        "longUrl": "https://www.google.org/"
    }
]

Send As API key

#!/usr/bin/python3.7

import asyncio, pprint
from aiogoogle import Aiogoogle

api_key = 'abc123'

async def translate_to_latin(words):
    async with Aiogoogle(api_key=api_key) as aiogoogle:
        language = await aiogoogle.discover('translate', 'v2')
        words = dict(q=[words], target='la')
        result = await aiogoogle.as_api_key(
            language.translations.translate(json=words)
        )
    pprint.pprint(result)

if __name__ == '__main__':
    asyncio.run(translate_to_latin('Aiogoogle is awesome'))

{
    "data": {
        "translations": [
            {
                "translatedText": "Aiogoogle est terribilis!",
                # Google probably meant "awesomelis", but whatever..
                "detectedSourceLanguage": "en"
            }
        ]
    }
}

Send As User

import asyncio
from aiogoogle import Aiogoole
from pprint import pprint

async def get_calendar_events():
    async with Aiogoogle(user_creds={'access_token': '...', 'refresh_token': '...'}) as aiogoogle:
        calendar_v3 = await aiogoogle.discover('calendar', 'v3')
        result = await aiogoogle.as_user(
            calendar_v3.events.list(
                calendarId="primary",
                maxResults=1
            )
        )
    pprint.pprint(result)

asyncio.run(get_calendar_events())

{
    "kind": "calendar#events",
    "etag": "\"p33c910kumb6ts0g\"",
    "summary": "user@gmail.com",
    "updated": "2018-11-11T22:31:03.463Z",
    "timeZone": "Africa/Cairo",
    "accessRole": "owner",
    "defaultReminders": [
        {
            "method": "popup",
            "minutes": 30
        },
        {
            "method": "email",
            "minutes": 30
        }
    ],
    "nextPageToken": "CigKGjVkcXIxa20wdHJrOW0xMXN0YABIAGNiQgp6yzd4C",
    "items": [
        {
            "kind": "calendar#event",
            "etag": "\"2784256013588000\"",
            "id": "asdasdasdasdasd",
            "status": "confirmed",
            "htmlLink": "https://www.google.com/calendar/event?eid=asdasdasdasdQG0",
            "created": "2014-02-11T14:13:26.000Z",
            "updated": "2014-02-11T14:13:26.794Z",
            "summary": "Do Something",
            "creator": {
                "email": "omarryhan@gmail.com",
                "displayName": "Omar Ryhan",
                "self": true
            },
            "organizer": {
                "email": "omarryhan@gmail.com",
                "displayName": "Omar Ryhan",
                "self": true
            },
            "start": {
                "date": "2014-03-07"
            },
            "end": {
                "date": "2014-03-08"
            },
            "iCalUID": "asdasdasd@google.com",
            "sequence": 0,
            "attendees": [
                {
                    "email": "omarryhan@gmail.com",
                    "displayName": "Omar Ryhan",
                    "organizer": true,
                    "self": true,
                    "responseStatus": "accepted"
                }
            ],
            "reminders": {
                "useDefault": false,
                "overrides": [
                    {
                        "method": "email",
                        "minutes": 450
                    },
                    {
                        "method": "popup",
                        "minutes": 30
                    }
                ]
            }
        }
    ]
}

Examples

You can find all examples in this directory. Some of them are listed here for convenience.

List Files on Google Drive

Full example here.

import asyncio
from aiogoogle import Aiogoogle

user_creds = {'access_token': '...', 'refresh_token': '...'}

async def list_files():
    async with Aiogoogle(user_creds=user_creds) as aiogoogle:
        drive_v3 = await aiogoogle.discover('drive', 'v3')
        json_res = await aiogoogle.as_user(
            drive_v3.files.list(),
        )
        for file in json_res['files']:
            print(file['name'])

asyncio.run(list_files())

Pagination

async def list_files():
    async with Aiogoogle(user_creds=user_creds) as aiogoogle:
        drive_v3 = await aiogoogle.discover('drive', 'v3')
        full_res = await aiogoogle.as_user(
            drive_v3.files.list(),
            full_res=True
        )
    async for page in full_res:
        for file in page['files']:
            print(file['name'])

asyncio.run(list_files())

Download a File from Google Drive

async def download_file(file_id, path):
    async with Aiogoogle(user_creds=user_creds) as aiogoogle:
        drive_v3 = await aiogoogle.discover('drive', 'v3')
        await aiogoogle.as_user(
            drive_v3.files.get(fileId=file_id, download_file=path, alt='media'),
        )
asyncio.run(download_file('abc123', '/home/user/Desktop/my_file.zip'))

Upload a File to Google Drive

async def upload_file(path):
    async with Aiogoogle(user_creds=user_creds) as aiogoogle:
        drive_v3 = await aiogoogle.discover('drive', 'v3')
        await aiogoogle.as_user(
            drive_v3.files.create(upload_file=path)
        )
asyncio.run(upload_file('/home/aiogoogle/Documents/my_cool_gif.gif/'))

Upload a File to Google Drive using an AsyncIterable

Full example here.

List Your Contacts

import asyncio

async def list_contacts():
    aiogoogle = Aiogoogle(user_creds=user_creds)
    people_v1 = await aiogoogle.discover('people', 'v1')
    async with aiogoogle:
        contacts_book = await aiogoogle.as_user(
            people_v1.people.connections.list(
                resourceName='people/me',
                personFields='names,phoneNumbers'
            ),
            full_res=True
        )
    async for page in contacts_book:
        for connection in page['connections']:
            print(connection['names'])
            print(connection['phoneNumbers'])

asyncio.run(list_contacts())

List Calendar Events

async def list_events():
    async with Aiogoogle(user_creds=user_creds) as aiogoogle:
        calendar_v3 = await aiogoogle.discover('calendar', 'v3')
        res = await aiogoogle.as_user(
            calendar_v3.events.list(calendarId='primary')
        )
        pprint.pprint(res)

Verify an Android In App Purchase

We disable raise_for_status here, to prevent raising an HTTP error on invalid purchases.

async def verify_purchase(token, package_name, product_id):
    async with Aiogoogle(
            service_account_creds=creds
    ) as aiogoogle:
        publisher_api = await aiogoogle.discover('androidpublisher', 'v3')

        request = publisher_api.purchases.products.get(
            token=token,
            productId=product_id,
            packageName=package_name)

        validation_result = await aiogoogle.as_service_account(
            request,
            full_res=True,
            raise_for_status=False
        )
    pprint(validation_result.content)

Check out https://github.com/omarryhan/aiogoogle/tree/master/examples for more.

Design and goals

Async framework agnostic

Aiogoogle does not and will not enforce the use of any async/await framework e.g. Asyncio, Curio or Trio. As a result, modules that handle io are made to be easily pluggable.

If you want to use Curio instead of Asyncio:

pip install aiogoogle[curio_asks]

e.g.

import curio

from aiogoogle import Aiogoogle
from aiogoogle.sessions.curio_asks_session import CurioAsksSession

async def main():
    async with Aiogoogle(session_factory=CurioAsksSession) as aiogoogle:
        youtube = await aiogoogle.discover('youtube', 'v3')

curio.run(main)

Another e.g.

#!/usr/bin/python3.7

import curio, pprint

from aiogoogle import Aiogoogle
from aiogoogle.sessions.curio_asks_session import CurioAsksSession

async def list_events():
    async with Aiogoogle(user_creds=user_creds, client_creds=client_creds, session_factory=CurioAsksSession) as aiogoogle:
        calendar_v3 = await aiogoogle.discover('calendar', 'v3')
        events = await aiogoogle.as_user(
            calendar_v3.events.list(calendarId='primary'),
        )
    pprint.pprint(events)

if __name__ == '__main__':
    curio.run(list_events)

The same with asyncio would look like this:

#!/usr/bin/python3.7

import asyncio, pprint

from aiogoogle import Aiogoogle
from aiogoogle.sessions.aiohttp_session import AiohttpSession  # Default

async def list_events():
    async with Aiogoogle(user_creds=user_creds, client_creds=client_creds, session_factory=AiohttpSession) as aiogoogle:
        calendar_v3 = await aiogoogle.discover('calendar', 'v3')
        events = await aiogoogle.as_user(
            calendar_v3.events.list(calendarId='primary'),
        )
    pprint.pprint(events)

if __name__ == '__main__':
    asyncio.run(list_events)

And Trio:

pip install aiogoogle[trio_asks]

#!/usr/bin/python3.7

import trio, pprint

from aiogoogle import Aiogoogle
from aiogoogle.sessions.trio_asks_session import TrioAsksSession   # Default

async def list_events():
    async with Aiogoogle(user_creds=user_creds, client_creds=client_creds, session_factory=TrioAsksSession) as aiogoogle:
        calendar_v3 = await aiogoogle.discover('calendar', 'v3')
        events = await aiogoogle.as_user(
            calendar_v3.events.list(calendarId='primary'),
        )
    pprint.pprint(events)

if __name__ == '__main__':
    trio.run(list_events)

Lightweight and minimalistic

Aiogoogle is built to be as lightweight and extensible as possible so that both client facing applications and API libraries can use it.

API

Aiogoogle

class aiogoogle.client.Aiogoogle(session_factory=<class 'aiogoogle.sessions.aiohttp_session.AiohttpSession'>, api_key=None, user_creds=None, client_creds=None, service_account_creds=None)[source]

Bases: object

Main entry point for Aiogoogle.

This class acts as tiny wrapper around:

Discovery Service v1 API

Aiogoogle’s OAuth2 manager

Aiogoogle’s API key manager

Aiogoogle’s OpenID Connect manager

Aiogoogle’s service account manager

One of Aiogoogle’s implementations of a session object

Parameters:

session_factory (aiogoogle.sessions.abc.AbstractSession) – AbstractSession Implementation. Defaults to aiogoogle.sessions.aiohttp_session.AiohttpSession
api_key (aiogoogle.auth.creds.ApiKey) – Google API key
user_creds (aiogoogle.auth.creds.UserCreds) – OAuth2 user credentials
client_creds (aiogoogle.auth.creds.ClientCreds) – OAuth2 client credentials
service_account_creds (aiogoogle.auth.creds.ServiceAccountCreds) – Service account credentials

Note

In case you want to instantiate a custom session with initial parameters, you can pass an anonymous factory. e.g.

>>> sess = lambda: Session(your_custome_arg, your_custom_kwarg=True)
>>> aiogoogle = Aiogoogle(session_factory=sess)

async as_anon(*requests, timeout=None, full_res=False, raise_for_status=True)[source]

Sends unauthorized requests

Parameters:

*requests (aiogoogle.models.Request) – Requests objects typically created by aiogoogle.resource.Method.__call__
timeout (int) – Total timeout for all the requests being sent
full_res (bool) – If True, returns full HTTP response object instead of returning it’s content
raise_for_status (bool) – If True, raises an HTTP error on HTTP status codes >= 400

Return type:

aiogoogle.models.Response

async as_api_key(*requests, timeout=None, full_res=False, api_key=None, raise_for_status=True)[source]

Sends requests on behalf of self.api_key (OAuth2)

Parameters:

*requests (aiogoogle.models.Request) – Requests objects typically created by aiogoogle.resource.Method.__call__
timeout (int) – Total timeout for all the requests being sent
full_res (bool) – If True, returns full HTTP response object instead of returning it’s content
api_key (string) – If you pass an API key here, it will only be used for this one request.
raise_for_status (bool) – If True, raises an HTTP error on HTTP status codes >= 400

Return type:

aiogoogle.models.Response

async as_service_account(*requests, timeout=None, full_res=False, service_account_creds=None, raise_for_status=True)[source]

Sends requests on behalf of self.user_creds (OAuth2)

Parameters:

*requests (aiogoogle.models.Request) – Requests objects typically created by aiogoogle.resource.Method.__call__
timeout (int) – Total timeout for all the requests being sent
full_res (bool) – If True, returns full HTTP response object instead of returning it’s content
service_account_creds (aiogoogle.auth.creds.ServiceAccountCreds) – You only have to pass service_account_creds once, either here or when instantiating an instance of Aiogoogle.
raise_for_status (bool) – If True, raises an HTTP error on HTTP status codes >= 400

Return type:

aiogoogle.models.Response

async as_user(*requests, timeout=None, full_res=False, user_creds=None, raise_for_status=True)[source]

Sends requests on behalf of self.user_creds (OAuth2)

Parameters:

*requests (aiogoogle.models.Request) – Requests objects typically created by aiogoogle.resource.Method.__call__
timeout (int) – Total timeout for all the requests being sent
full_res (bool) – If True, returns full HTTP response object instead of returning it’s content
user_creds (aiogoogle.auth.creds.UserCreds) – If you pass user_creds here, they will only be used for this one request.
raise_for_status (bool) – If True, raises an HTTP error on HTTP status codes >= 400

Return type:

aiogoogle.models.Response

async discover(api_name, api_version=None, validate=False, *, disco_doc_ver: int | None = None)[source]

Donwloads a discovery document from Google’s Discovery Service V1 and sets it a aiogoogle.resource.GoogleAPI

Note

It is recommended that you explicitly specify an API version.

When you leave the API version as None, Aiogoogle uses the list_api method to search for the best fit version of the given API name.

This will result in sending two http requests instead of just one.

Parameters:

api_name (str) – API name to discover. e.g.: “youtube”
api_version (str) – API version to discover e.g.: “v3” not “3” and not 3
validate (bool) – Set this to True to use this lib’s built in parameter validation logic. Note that you shouldn’t rely on this for critical user input validation.
disco_doc_ver – Specify which Google Discovery Service version to fetch a discovery document with. Useful for fetching discovery docs for Google APIs that aren’t supported by the default version of the Google Discovery Service Aiogoogle uses.

Returns:

An object that will then be used to create API requests

Return type:

aiogoogle.resource.GoogleAPI

Raises:

aiogoogle.excs.HTTPError –

async list_api(name, preferred=None, fields=None)[source]

https://developers.google.com/discovery/v1/reference/apis/list

The discovery.apis.list method returns the list all APIs supported by the Google APIs Discovery Service.

The data for each entry is a subset of the Discovery Document for that API, and the list provides a directory of supported APIs.

If a specific API has multiple versions, each of the versions has its own entry in the list.

Example

>>> await aiogoogle.list_api('youtube')

{
    "kind": "discovery#directoryList",
    "discoveryVersion": "v1",
    "items": [
        {
            "kind": "discovery#directoryItem",
            "id": "youtube:v3",
            "name": "youtube",
            "version": "v3",
            "title": "YouTube Data API",
            "description": "Supports core YouTube features, such as uploading videos, creating and managing playlists, searching for content, and much more.",
            "discoveryRestUrl": "https://www.googleapis.com/discovery/v1/apis/youtube/v3/rest",
            "discoveryLink": "./apis/youtube/v3/rest",
            "icons": {
                "x16": "https://www.google.com/images/icons/product/youtube-16.png",
                "x32": "https://www.google.com/images/icons/product/youtube-32.png"
            },
            "documentationLink": "https://developers.google.com/youtube/v3",
            "preferred": true
        }
    ]
}

Parameters:

name (str) – Only include APIs with the given name.
preferred (bool) – Return only the preferred version of an API. “false” by default.
fields (str) – Selector specifying which fields to include in a partial response.

Return type:

dict

Raises:

aiogoogle.excs.HTTPError –

async send(*args, **kwargs)[source]

GoogleAPI

class aiogoogle.resource.GoogleAPI(discovery_document, validate=False)

Bases: object

Creetes a representation of Google API given a discovery document

Parameters:

discovery_document (dict) – A discovery document
validate (bool) – Set this to True to use this lib’s built in parameter validation logic. Note that you shouldn’t rely on this for critical user input validation.

__getattr__(method_or_resource) → Resource

Returns resources from an API

Note

This method will first check in resources then will check in methods.

Parameters:: method_or_resource (str) – name of the top level method or resource

Example

>>> google_service = GoogleAPI(google_service_discovery_doc)
>>> google_service.user

user resource @ google_service.com/api/v1/

Returns:: A Resource or a Method
Return type:: aiogoogle.resource.Resource, aiogoogle.resource.Method
Raises:: AttributeError –

__getitem__(k)

Returns items from the discovery document

Example

>>> google_service = GoogleAPI(google_books_discovery_doc)
>>> google_service['name']

'books'

>>> google_service['id']

'books:v1'

>>> google_service['version']

'v1'

>>> google_service['documentationLink']

'https://developers.google.com/books/docs/v1/getting_started'

>>> google_service['oauth2']['scopes']

https://www.googleapis.com/auth/books: {

    description: "Manage your books"

}

Returns:: Discovery Document Item
Return type:: dict

property methods_available: List[str]: Returns names of the methods provided by this resource

property resources_available: List[str]: Returns names of the resources in a given API if any

Resource

class aiogoogle.resource.Resource(name, resource_specs, global_parameters, schemas, root_url, service_path, batch_path, validate)

Bases: object

__getattr__(method_or_resource)

Returns either a method or a nested resource

Parameters:: method_or_resource – Name of the method or resource desired.
Returns:: A Resource or a Method
Return type:: aiogoogle.resource.Resource, aiogoogle.resource.Methods

Note

This method will first check in nested resources then will check in methods.

Raises:: AttributeError –

property methods_available: List[str]: Returns the names of the methods that this resource provides

property resources_available: List[str]: Returns the names of the nested resources in this resource

Method

class aiogoogle.resource.Method(name, method_specs, global_parameters, schemas, root_url, service_path, batch_path, validate)

Bases: object

__call__(validate=None, data=None, json=None, upload_file=None, pipe_from=None, download_file=None, pipe_to=None, timeout=None, path_params_safe_chars={}, **uri_params) → Request

Builds a request from this method

Note

When passing datetime.datetime or datetime.date pass them in json format.
Aiogoogle won’t do that as it would be a big hassle to iterate over every item in *uri_params, json and data to check if there’s any datetime objects.
Fortunately Python makes it really easy to achieve that.
Instead of passing say datetime.datetime.utcnow(), pass: datetime.datetime.utcnow().jsonformat()

Note

All None values are ommited before sending to Google apis, if you want to explicitly pass a JSON null then pass it as "null" not None

Parameters:

validate (bool) – Overrides :param: aiogoogle.Aiogoole.validate if not None
json (dict) – Json body
data (any) – Data body (Bytes, text, www-url-form-encoded and others)
upload_file (str) – full path of file to upload
download_file (str) – full path of file to download to
timeout (str) – total timeout for this request
path_params_safe_chars (dict) – Dictionary of safe characters for each path parameter.
**uri_params (dict) – path and query, required and optional parameters

Returns:

An unsent request object

Return type:

aiogoogle.models.Request

__getitem__(key)

Examples

>>> self['description']

"method description"

>>> self['scopes']

['returns', 'scopes', 'required', 'by', 'this', 'method', 'in', 'a', 'list']

>>> self['supportsMediaDownload']

False

>>> self['supportsMediaUpload']

True

>>> self['httpMethod']

'GET'

Hint

Using this method with scopes as an argument can be useful for incremental authorization. (Requesting scopes when needed. As opposed to requesting them at once)

for more: https://developers.google.com/identity/protocols/OAuth2WebServer#incrementalAuth

property optional_parameters: List[str]

Optional Parameters

Returns:: List of the names of optional parameters this method takes
Return type:: list

property optional_query_parameters: List[str]

Optional Query Parameters

Returns:: List of the names of optional query parameters this method takes
Return type:: list

property parameters: dict

Parameters property

Returns:: All parameters that this method can take as described in the discovery document
Return type:: dict

property path_parameters: List[str]

Path Parameters

Returns:: List of the names of path parameters this method takes
Return type:: list

property query_parameters: List[str]

Query Parameters

Returns:: List of the names of Query parameters this method takes
Return type:: list

property request: dict: Returns expected request body

property required_parameters: List[str]

Required Parameters

Returns:: List of the names of required parameters this method takes
Return type:: list

property required_query_parameters: List[str]

Required Query Parameters

Returns:: List of the names of required query parameters this method takes
Return type:: list

property response: dict: Retruns expected response body

Credentials

class aiogoogle.auth.creds.ApiKey[source]: Bases: str

class aiogoogle.auth.creds.ClientCreds(client_id=None, client_secret=None, scopes=None, redirect_uri=None)[source]

Bases: _dict

OAuth2 Client Credentials Dictionary

Examples

Scopes: [‘openid’, ‘email’, ‘https://www.googleapis.com/auth/youtube.force-ssl’]

client_id

OAuth2 client ID

Type:: str

client_secret

OAuth2 client secret

Type:: str

scopes

List of scopes that this client should request from resource server

Type:: list

redirect_uri

client’s redirect uri

Type:: str

class aiogoogle.auth.creds.ServiceAccountCreds(type=None, project_id=None, private_key_id=None, private_key=None, client_email=None, client_id=None, auth_uri=None, token_uri=None, auth_provider_x509_cert_url=None, client_x509_cert_url=None, subject=None, scopes=None, additional_claims=None)[source]

Bases: _dict

Service account key (the one you download from Google Cloud Console) + some additional optional attributes. If you have created the service account using Google’s IAM REST API, rather than downloading it using gcloud or Google’s cloud console, the downloaded JSON key file will look different. Check here: https://cloud.google.com/iam/docs/creating-managing-service-account-keys#creating_service_account_keys to be able to pass the correct info to this data model.

type

Should be “service_account”

Type:: string

project_id

Your project ID

Type:: string

private_key_id

Private key ID

Type:: string

private_key

RSA Private key

Type:: string

client_email

service account email

Type:: string

client_id

service account ID

Type:: string

auth_uri

Auth URI

Type:: string

token_uri

Token URI

Type:: string

auth_provider_x509_cert_url

Auth provider cert URL

Type:: string

client_x509_cert_url

Cert URL

Type:: string

scopes

Scopes to request during the authorization grant.

Type:: Sequence[str]

subject

For domain-wide delegation, the email address of the user to for which to request delegated access.

Type:: str

additional_claims

Any additional claims for the JWT assertion used in the authorization grant.

Type:: Mapping[str, str]

Example

service_account_creds = ServiceAccountCreds(
    scopes=["https://www.googleapis.com/auth/cloud-platform"],
    **json.load(open('service-account-key.json'))
)

# or

service_account_creds = {
    "scopes": ["..."],
    **json.load(open('service-account-key.json'))
}

class aiogoogle.auth.creds.UserCreds(access_token=None, refresh_token=None, expires_in=None, expires_at=None, scopes=None, id_token=None, id_token_jwt=None, token_type=None, token_uri=None, token_info_uri=None, revoke_uri=None)[source]

Bases: _dict

OAuth2 User Credentials Dictionary

access_token

Access Token

Type:: str

refresh_token

Refresh Token

Type:: str

expires_in

seconds till expiry from creation

Type:: int

expires_at

JSON datetime ISO 8601 expiry datetime

Type:: str

scopes

list of scopes owned by access token

Type:: list

id_token

Decoded OpenID JWT

Type:: aiogoogle.auth.creds.IdToken

id_token_jwt

Encoded OpenID JWT

Type:: str

token_type

Bearer

Type:: str

token_uri

URI where this token was issued from

Type:: str

token_info_uri

URI where one could get more info about this token

Type:: str

revoke_uri

URI where this token should be revoked

Type:: str

Auth Managers

Note

Aiogoogle is an async-framework agnosting library. As a result, adding file IO OR credentials storage capabilities for user credentials would add unneeded complexity.
Credentials are an instance of dict and are guaranteed to only contain JSON types (str, number, array, JSONSCHEMA datetime and ISO8601 datetime, etc) to make them easily serializable.

class aiogoogle.auth.managers.ApiKeyManager(api_key=None)[source]

Bases: object

authorize(request, key=None) → Request[source]

Adds API Key authorization query argument to URL of a request given an API key

Parameters:

request (aiogoogle.models.Request) – Request to authorize
creds (aiogoogle.auth.creds.ApiKey) – ApiKey to refresh with

Returns:

Request with API key in URL

Return type:

aiogoogle.models.Request

class aiogoogle.auth.managers.Oauth2Manager(session_factory=<class 'aiogoogle.sessions.aiohttp_session.AiohttpSession'>, client_creds=None)[source]

Bases: object

OAuth2 manager that only supports Authorization Code Flow (https://tools.ietf.org/html/rfc6749#section-1.3.1)

Parameters:

session_factory (aiogoogle.sessions.AbstractSession) – A session implementation
verify (bool) – whether or not to verify tokens fetched

__getitem__(key)[source]

Gets Google’s openID configs

Example

response_types_supported
scopes_supported
claims_supported

authorization_url(client_creds=None, state=None, access_type=None, include_granted_scopes=None, login_hint=None, prompt=None, response_type='code', scopes=None) → str[source]

First step of OAuth2 authoriztion code flow. Creates an OAuth2 authorization URI.

Parameters:

client_creds (aiogoogle.auth.creds.ClientCreds) –
A client_creds object/dictionary containing the following items:
- client_id
- scopes
- redirect_uri
scopes (list) –
List of OAuth2 scopes to ask for
- Optional
- Overrides the list of scopes specified in client creds
state (str) –
A CSRF token
- Optional
- Specifies any string value that your application uses to maintain state between your authorization request and the authorization server’s response.
- The server returns the exact value that you send as a name=value pair in the hash (#) fragment of the redirect_uri after the user consents to or denies your application’s access request.
- You can use this parameter for several purposes, such as:
  - Directing the user to the correct resource in your application
  - Sending nonces
  - Mitigating cross-site request forgery.
- If no state is passed, this method will generate and add a secret token to user_creds['state'].
- Since your redirect_uri can be guessed, using a state value can increase your assurance that an incoming connection is the result of an authentication request.
- If you generate a random string or encode the hash of a cookie or another value that captures the client’s state, you can validate the response to additionally ensure that the request and response originated in the same browser, providing protection against attacks such as cross-site request forgery.
access_type (str) –
Indicates whether your application can refresh access tokens when the user is not present at the browser. Options:
- Optional
- "online" Default
- "offline" Choose this for a refresheable/long-term access token
include_granted_scopes (bool) –
- Optional
- Enables applications to use incremental authorization to request access to additional scopes in context.
- If you set this parameter’s value to True and the authorization request is granted, then the new access token will also cover any scopes to which the user previously granted the application access.
login_hint (str) –
- Optional
- If your application knows which user is trying to authenticate, it can use this parameter to provide a hint to the Google Authentication Server.
- The server uses the hint to simplify the login flow either by prefilling the email field in the sign-in form or by selecting the appropriate multi-login session.
- Set the parameter value to an email address or sub identifier, which is equivalent to the user’s Google ID.
- This can help you avoid problems that occur if your app logs in the wrong user account.
prompt (str) –
- Optional
- A space-delimited, case-sensitive list of prompts to present the user.
- If you don’t specify this parameter, the user will be prompted only the first time your app requests access.
- Possible values are:
  - None : Default: Do not display any authentication or consent screens. Must not be specified with other values.
  - 'consent' : Prompt the user for consent.
  - 'select_account' : Prompt the user to select an account.
response_type (str) –
- Optional
- OAuth2 response type
- Defaults to Authorization Code Flow response type

Note

It is highly recommended that you don’t leave state as None in production.
To effortlessly create a random secret to pass it as a state token, you can use aiogoogle.auth.utils.create_secret()

Note

A Note About Scopes:

For a list of all of Google’s available scopes: https://developers.google.com/identity/protocols/googlescopes
It is recommended that your application requests access to authorization scopes in context whenever possible.
By requesting access to user data in context, via incremental authorization, you help users to more easily understand why your application needs the access it is requesting.

Warning

When listening for a callback after redirecting a user to the URL returned from this method, take the following into consideration:
- If your response endpoint renders an HTML page, any resources on that page will be able to see the authorization code in the URL.
- Scripts can read the URL directly, and the URL in the Referer HTTP header may be sent to any or all resources on the page.
- Carefully consider whether you want to send authorization credentials to all resources on that page (especially third-party scripts such as social plugins and analytics).
- To avoid this issue, it’s recommend that the server first handle the request, then redirect to another URL that doesn’t include the response parameters.

Example

from aiogoogle.auth.utils import create_secret
from aiogoogle import ClinetCreds

client_creds = ClientCreds(
    client_id='a_client_id',
    scopes=['first.scope', 'second.scope'],
    redirect_uri='http://localhost:8080'
)

state = create_secret()

auth_uri = oauth2.authorization_url(
    client_creds=client_creds,
    state=state,
    access_type='offline',
    include_granted_scopes=True,
    login_hint='example@gmail.com',
    prompt='select_account'
    )

Returns:: An Authorization URI
Return type:: (str)

static authorize(request: Request, user_creds: dict) → Request[source]

Adds OAuth2 authorization headers to requests given user creds

Parameters:

request (aiogoogle.models.Request) – Request to authorize
user_creds (aiogoogle.auth.creds.UserCreds) – user_creds to refresh with

Returns:

Request with OAuth2 authorization header

Return type:

aiogoogle.models.Request

static authorized_for_method(method, user_creds) → bool[source]

Checks if oauth2 user_creds dict has sufficient scopes for a method call.

Note

This method doesn’t check whether creds are refreshed or valid.

e.g.

Correct:

is_authorized = authorized_for_method(youtube.video.list)

NOT correct:

is_authorized = authorized_for_method(youtube.video.list())

AND NOT correct:

is_authorized = authorized_for_method(youtube.videos)

Parameters:

method (aiogoogle.resource.Method) – Method to be checked
user_credentials (aiogoogle.auth.creds.UserCreds) – User Credentials with scopes item

Return type:

bool

async build_user_creds(grant, client_creds=None, grant_type='authorization_code') → dict[source]

Second step of Oauth2 authrization code flow. Creates a User Creds object with access and refresh token

Parameters:

grant (str) –
- Aka: “code”.
- The code received at your redirect URI from the auth callback
client_creds (aiogoogle.auth.creds.ClientCreds) –
- Dict with client_id and client_secret items
grant_type (str) –
- Optional
- OAuth2 grant type
- defaults to code (Authorization code flow)

Returns:

User Credentials with the following items:

access_token

refresh_token

expires_in (JSON format ISO 8601)

token_type always set to bearer

scopes

Return type:

aiogoogle.auth.creds.UserCreds

Raises:

aiogoogle.excs.AuthError – Auth Error

async get_me_info(user_creds)[source]

Gets information of a user given his access token. User must also be the client. (Not sure whether or not that’s the main purpose of this endpoint and how it differs from get_user_info. If someone can confirm/deny the description above, please edit (or remove) this message and make a pull request)

Parameters:: user_creds (aiogoogle.creds.UserCreds) – UserCreds instance with an access token
Returns:: Info about the user
Return type:: dict
Raises:: aiogoogle.excs.HTTPError –

async get_token_info(user_creds)[source]

Gets token info given an access token

Parameters:: user_creds (aiogoogle.creds.UserCreds) – UserCreds instance with an access token
Returns:: Info about the token
Return type:: dict

static is_expired(creds) → bool[source]

Checks if user_creds expired

Parameters:: user_creds (aiogoogle.auth.creds.UserCreds) – User Credentials
Return type:: bool

is_ready(client_creds=None)[source]

Checks passed client_creds whether or not the client has enough information to perform OAuth2 Authorization code flow

Parameters:: client_creds (aiogoogle.auth.creds.ClientCreds) – Client credentials object
Return type:: bool

async refresh(user_creds, client_creds=None)[source]

Refreshes user_creds

Parameters:

user_creds (aiogoogle.auth.creds.UserCreds) – User Credentials with refresh_token item
client_creds (aiogoogle.auth.creds.ClientCreds) – Client Credentials

Returns:

If the token is refreshed or not aiogoogle.creds.UserCreds: Refreshed user credentials

Return type:

bool

Raises:

aiogoogle.excs.AuthError – Auth Error

async revoke(user_creds)[source]

Revokes user_creds

In some cases a user may wish to revoke access given to an application. A user can revoke access by visiting Account Settings. It is also possible for an application to programmatically revoke the access given to it. Programmatic revocation is important in instances where a user unsubscribes or removes an application. In other words, part of the removal process can include an API request to ensure the permissions granted to the application are removed.

Parameters:: user_creds (aiogoogle.auth.Creds) – UserCreds with an access_token item
Return type:: None
Raises:: aiogoogle.excs.AuthError –

class aiogoogle.auth.managers.OpenIdConnectManager(*args, **kwargs)[source]

Bases: Oauth2Manager

authorization_url(client_creds=None, nonce=None, state=None, prompt=None, display=None, login_hint=None, access_type=None, include_granted_scopes=None, openid_realm=None, hd=None, response_type='code', scopes=None)[source]

First step of OAuth2 authoriztion code flow. Creates an OAuth2 authorization URI.

Parameters:

client_creds (aiogoogle.auth.creds.ClientCreds) –
A client_creds object/dictionary containing the following items:
- client_id
- scopes
  - The scope value must begin with the string openid and then include profile or email or both.
- redirect_uri
nonce (str) – A random value generated by your app that enables replay protection.
scopes (list) –
List of OAuth2 scopes to ask for
- Optional
- Overrides the list of scopes specified in client creds
- Some OpenID scopes that you can include: [‘email’, ‘profile’, ‘openid’]
display (str) –
- Optional
- An ASCII string value for specifying how the authorization server displays the authentication and consent user interface pages.
- The following values are specified, and accepted by the Google servers, but do not have any effect on its behavior:
  - page
  - popup
  - touch
  - wap
state (str) –
A CSRF token
- Optional
- Specifies any string value that your application uses to maintain state between your authorization request and the authorization server’s response.
- The server returns the exact value that you send as a name=value pair in the hash (#) fragment of the redirect_uri after the user consents to or denies your application’s access request.
- You can use this parameter for several purposes, such as:
  - Directing the user to the correct resource in your application
  - Sending nonces
  - Mitigating cross-site request forgery.
- If no state is passed, this method will generate and add a secret token to user_creds['state'].
- Since your redirect_uri can be guessed, using a state value can increase your assurance that an incoming connection is the result of an authentication request.
- If you generate a random string or encode the hash of a cookie or another value that captures the client’s state, you can validate the response to additionally ensure that the request and response originated in the same browser, providing protection against attacks such as cross-site request forgery.
access_type (str) –
Indicates whether your application can refresh access tokens when the user is not present at the browser. Options:
- Optional
- "online" Default
- "offline" Choose this for a refresheable/long-term access token
include_granted_scopes (bool) –
- Optional
- Enables applications to use incremental authorization to request access to additional scopes in context.
- If you set this parameter’s value to True and the authorization request is granted, then the new access token will also cover any scopes to which the user previously granted the application access.
login_hint (str) –
- Optional
- If your application knows which user is trying to authenticate, it can use this parameter to provide a hint to the Google Authentication Server.
- The server uses the hint to simplify the login flow either by prefilling the email field in the sign-in form or by selecting the appropriate multi-login session.
- Set the parameter value to an email address or sub identifier, which is equivalent to the user’s Google ID.
- This can help you avoid problems that occur if your app logs in the wrong user account.
prompt (str) –
- Optional
- A space-delimited, case-sensitive list of prompts to present the user.
- If you don’t specify this parameter, the user will be prompted only the first time your app requests access.
- Possible values are:
  - None : Default: Do not display any authentication or consent screens. Must not be specified with other values.
  - 'consent' : Prompt the user for consent.
  - 'select_account' : Prompt the user to select an account.
openid_realm (str) –
- Optional
- openid.realm is a parameter from the OpenID 2.0 protocol.
- It is used in OpenID 2.0 requests to signify the URL-space for which an authentication request is valid.
- Use openid.realm if you are migrating an existing application from OpenID 2.0 to OpenID Connect.
- For more details, see Migrating off of OpenID 2.0. https://developers.google.com/identity/protocols/OpenID2Migration
hd (str) –
- Optional
- The hd (hosted domain) parameter streamlines the login process for G Suite hosted accounts.
- By including the domain of the G Suite user (for example, mycollege.edu), you can indicate that the account selection UI should be optimized for accounts at that domain.
- To optimize for G Suite accounts generally instead of just one domain, use an asterisk: hd=*.
- Don’t rely on this UI optimization to control who can access your app, as client-side requests can be modified.
- Be sure to validate that the returned ID token has an hd claim value that matches what you expect (e.g. mycolledge.edu).
- Unlike the request parameter, the ID token claim is contained within a security token from Google, so the value can be trusted.
response_type (str) –
- Optional
- OAuth2 response type
- Defaults to Authorization Code Flow response type

Note

It is highly recommended that you don’t leave state as None in production

To effortlessly create a random secret to pass it as a state token, you can use aiogoogle.auth.utils.create_secret()

Note

A Note About Scopes:

You can mix OAuth2 scopes with OpenID connect scopes. e.g.: openid email https://www.googleapis.com/auth/urlshortener
For a list of all of Google’s available scopes: https://developers.google.com/identity/protocols/googlescopes
It is recommended that your application requests access to authorization scopes in context whenever possible.
By requesting access to user data in context, via incremental authorization, you help users to more easily understand why your application needs the access it is requesting.

Warning

When listening for a callback after redirecting a user to the URL returned from this method, take the following into consideration:
- If your response endpoint renders an HTML page, any resources on that page will be able to see the authorization code in the URL.
- Scripts can read the URL directly, and the URL in the Referer HTTP header may be sent to any or all resources on the page.
- Carefully consider whether you want to send authorization credentials to all resources on that page (especially third-party scripts such as social plugins and analytics).
- To avoid this issue, it’s recommend that the server first handle the request, then redirect to another URL that doesn’t include the response parameters.

Example

from aiogoogle.auth.utils import create_secret
from aiogoogle import ClinetCreds

client_creds = ClientCreds(
    client_id='a_client_id',
    scopes=['first.scope', 'second.scope'],
    redirect_uri='http://localhost:8080'
)

state = create_secret()
nonce = create_secret()

auth_uri = openid_connect.authorization_url(
    client_creds=client_creds,
    nonce=nonce,
    state=state,
    access_type='offline',
    include_granted_scopes=True,
    login_hint='example@gmail.com',
    prompt='select_account'
    )

Returns:: An Authorization URI
Return type:: (str)

async build_user_creds(grant, client_creds=None, grant_type='authorization_code', nonce=None, hd=None, verify=True)[source]

Second step of Oauth2 authrization code flow and OpenID connect. Creates a User Creds object with access and refresh token

Parameters:

grant (str) –
- Aka: “code”.
- The code received at your redirect URI from the auth callback
client_creds (aiogoogle.auth.creds.ClientCreds) –
- Dict with client_id and client_secret items
grant_type (str) –
- Optional
- OAuth2 grant type
- defaults to code (Authorization code flow)
nonce (str) –
- Random value that prevents replay attacks
- pass the one you used with self.authorization_url() method
hd (str) –
- Optional
- hosted domain for G-suite
- used for id_token verification
verify (str) –
- Optional
- Whether or not to verify the received id_token
- Unless you’re building a speed critical application AND you’re receiving your tokens directly from Google, you should leave this as True.

Returns:

User Credentials with the following items:

access_token

refresh_token

expires_in (JSON format ISO 8601)

token_type always set to bearer

scopes

Return type:

aiogoogle.auth.creds.UserCreds

Raises:

aiogoogle.excs.AuthError – Auth Error

async decode_and_validate(id_token_jwt, client_id=None, nonce=None, hd=None)[source]

Decodes then validates an openid_connect jwt with Google’s oaauth2 certificates

Parameters:

id_token_jwt (str) – Found in :class:aiogoogle.auth.creds.UserCreds
client_id (str) – If provided will validate token’s audience (‘aud’)
nonce (str) – If provided, will validate the nonce provided at authorization
hd (str) – If provided, will validate client’s hosted domain

Returns:

Decoded OpenID connect JWT

Return type:

dict

async get_token_info_jwt(user_creds)[source]

get token info using id_token_jwt instead of access_token self.get_token_info

Parameters:: user_creds (aiogoogle.auth.creds.UserCreds) – user_creds with id_token_jwt item
Returns:: Information about the token
Return type:: dict

async get_user_info(user_creds)[source]

https://developers.google.com/+/web/api/rest/openidconnect/getOpenIdConnect

People: getOpenIdConnect

Get user information after performing an OpenID connect flow.

Use this method instead of people.get (Google+ API) when you need the OpenID Connect format.

This method is not discoverable nor is it in the Google API client libraries.

To learn more, see OpenID Connect for sign-in. https://developers.google.com/+/web/api/rest/openidconnect/index.html

Example

>>> await get_user_info(user_creds)
{
    "kind": "plus#personOpenIdConnect",
    "gender": string,
    "sub": string,
    "name": string,
    "given_name": string,
    "family_name": string,
    "profile": string,
    "picture": string,
    "email": string,
    "email_verified": "true",
    "locale": string,
    "hd": string
}

Parameters:: user_creds (aiogoogle.auth.creds.UserCreds) – User credentials

class aiogoogle.auth.managers.ServiceAccountManager(session_factory=<class 'aiogoogle.sessions.aiohttp_session.AiohttpSession'>, creds=None)[source]

Bases: object

Parameters:

session_factory (aiogoogle.sessions.AbstractSession) – A session implementation
creds (aiogoogle.auth.creds.ServiceAccountCreds) – Service account creds

authorize(request, access_token=None)[source]

Adds the access token generated from the refresh method to the request’s header

Parameters:

request (aiogoogle.models.Request) – Request to authorize
access_token (string) – Optional access token

Returns:

Request with an OAuth2 bearer access token

Return type:

aiogoogle.models.Request

async detect_default_creds_source()[source]

Detects the most suitable method of service account authorization.

No need to call this method if you’ve already passed a service account private key to this object, i.e. Aiogoogle(service_account_creds={"private_key": "..."}) or Aiogoogle(service_account_creds=json.load(open('service_account_key.json')) If you didn’t, then you should ideally call it only once, after you instantiate this object.

Should follow Google’s default sequence (Loads creds from the first one that succeeds):

Loads service account creds from environment “GOOGLE_APPLICATION_CREDENTIALS”
Google Cloud SDK default credentials (Not yet available)
Google App Engine default credentials (Not yet available)
Compute Engine default credentials

Returns:: None

async refresh()[source]

Ensures that there’s an unexpired access token.

Returns:: If the token is refreshed or not.
Return type:: bool

Exceptions

exception aiogoogle.excs.AiogoogleError[source]: Bases: Exception

exception aiogoogle.excs.AuthError(msg, req=None, res=None)[source]: Bases: HTTPError

exception aiogoogle.excs.HTTPError(msg, req=None, res=None)[source]: Bases: AiogoogleError

exception aiogoogle.excs.ValidationError[source]

Bases: AiogoogleError

Raised when the validate flag is set true and a validation error occurs

Models

class aiogoogle.models.MediaDownload(file_path=None, chunk_size=None, pipe_to=None)[source]

Bases: object

Media Download

Parameters:

file_path (str) – Full path of the file to be downloaded
chunksize (int) – Size of a chunk of bytes that a session should write at a time when downloading.
pipe_to (object) – class object to stream file content to

class aiogoogle.models.MediaUpload(file_path_or_bytes, upload_path=None, mime_range=None, max_size=None, multipart=False, chunk_size=None, resumable=None, validate=True, pipe_from=None)[source]

Bases: object

Media Upload

Parameters:

file_path_or_bytes (str, bytes) – Full path or content of the file to be uploaded
upload_path (str) – The URI path to be used for upload. Should be used in conjunction with the rootURL property at the API-level.
mime_range (list) – list of MIME Media Ranges for acceptable media uploads to this method.
max_size (int) – Maximum size of a media upload in bytes
multipart (bool) – True if this endpoint supports upload multipart media.
chunksize (int) – Size of a chunk of bytes that a session should read at a time when uploading in multipart.
resumable (aiogoogle.models.ResumableUplaod) – A ResumableUpload object
validate (bool) – Whether or not a session should validate the upload size before sending
pipe_from (file object, AsyncIterable) – class object to stream file content from

async aiter_file(aiter_func)[source]

async read_file(read_func)[source]

async run_validation(size_func)[source]

class aiogoogle.models.Request(method=None, url=None, batch_url=None, headers=None, json=None, data=None, media_upload=None, media_download=None, timeout=None, callback=None, _verify_ssl=True, upload_file_content_type=None)[source]

Bases: object

Request class for the whole library. Auth Managers, GoogleAPI and Sessions should all use this.

Note

For HTTP body, only pass one of the following params:

json: json as a dict

data: www-url-form-encoded form as a dict/ bytes/ text/

Parameters:

method (str) – HTTP method as a string (upper case) e.g. ‘GET’
url (str) – full url as a string. e.g. ‘https://example.com/api/v1/resource?filter=filter#something
batch_url (str) – full url of for sending this request in a batch
json (dict) – json as a dict
data (any) – www-url-form-encoded form as a dict/ bytes/ text/
headers (dict) – headers as a dict
media_download (aiogoogle.models.MediaDownload) – MediaDownload object
media_upload (aiogoogle.models.MediaUpload) – MediaUpload object
timeout (int) – Individual timeout for this request
callback (callable) – Synchronous callback that takes the content of the response as the only argument. Should also return content.
_verify_ssl (boolean) – Defaults to True.
upload_file_content_type (str) – Optional content-type header string. In case you don’t want to use the default application/octet-stream (Or whatever is auto-detected by your transport handler)

classmethod batch_requests(*requests)[source]

Given many requests, will create a batch request per https://developers.google.com/discovery/v1/batch

Parameters:: *requests (aiogoogle.models.Request) – Request objects
Return type:: aiogoogle.models.Request

classmethod from_response(response)[source]

class aiogoogle.models.Response(status_code=None, headers=None, url=None, json=None, data=None, reason=None, req=None, download_file=None, pipe_to=None, upload_file=None, pipe_from=None, session_factory=None, auth_manager=None, user_creds=None)[source]

Bases: object

Respnse Object

Parameters:

status_code (int) – HTTP Status code
headers (dict) – HTTP response headers
url (str) – Request URL
json (dict) – Json Response if any
data (any) – data
reason (str) – reason for http error if any
req (aiogoogle.models.Request) – request that caused this response
download_file (str) – path of the download file specified in the request
pipe_to (object) – class object to stream file content to specified in the request.
upload_file (str) – path of the upload file specified in the request
pipe_from (file object) – class object to stream file content from
session_factory (aiogoogle.sessions.abc.AbstractSession) – A callable implementation of aiogoogle’s session interface
auth_manager (aiogoogle.auth.managers.ServiceAccountManager) – Service account authorization manager.
user_creds (aiogoogle.auth.creds.UserCreds) – user_creds to make an api call with.

__call__(session_factory=None, req_token_name=None, res_token_name=None, json_req=False)[source]

Returns a generator that yields the contents of the next pages if any (and this page as well)

Parameters:

session_factory (aiogoogle.sessions.abc.AbstractSession) – A session factory
req_token_name (str) –
- name of the next_page token in the request
- Default: “pageToken”
res_token_name (str) –
- name of the next_page token in json response
- Default: “nextPageToken”
json_req (dict) – Normally, nextPageTokens should be sent in URL query params. If you want it in A json body, set this to True

Returns:

self._next_page_generator (staticmethod)

Return type:

async generator

property content: Equals either self.json or self.data

property error_msg

next_page(req_token_name=None, res_token_name=None, json_req=False) → Request[source]

Method that returns a request object that requests the next page of a resource

Parameters:

req_token_name (str) –
- name of the next_page token in the request
- Default: “pageToken”
res_token_name (str) –
- name of the next_page token in json response
- Default: “nextPageToken”
json_req (dict) – Normally, nextPageTokens should be sent in URL query params. If you want it in A json body, set this to True

Return type:

A request object (aiogoogle.models.Request)

raise_for_status()[source]

class aiogoogle.models.ResumableUpload(multipart=None, chunk_size=None, upload_path=None)[source]

Bases: object

Resumable Upload Object. Works in conjuction with media upload

Parameters:

file_path (str) – Full path of the file to be uploaded
upload_path (str) – The URI path to be used for upload. Should be used in conjunction with the rootURL property at the API-level.
multipart (bool) – True if this endpoint supports upload multipart media.
chunk_size (int) – Size of a chunk of bytes that a session should read at a time when uploading in multipart.

Utils

aiogoogle.auth.utils.create_secret(bytes_length=1024)[source]

Validate

A simple instance validation module for Discovery schemas. Unfrtunately, Google uses a slightly modified version of JSONschema draft3 (mix of jsonschema and protobuf). As a result, using an external library to validate Discovery schemas will raise lots of errors. I tried to modify the popular: https://github.com/Julian/jsonschema to make it work with Google’s version, but it was just too complicated for the relatively simple task on our hands. If you face any problems with aiogoogle’s validation, you can always turn it off by either passing: method.__call__(validate=False) or turning it off for the whole API by passing aiogoogle.discover(validate=False)

This module misses a lot of the features provided with more advanced jsonschema validators e.g.

collecting all validation errors and raising them all at once
more descriptive errors with nice tracebacks
Validating repeatable instances
additionalItems

and more: https://tools.ietf.org/html/draft-zyp-json-schema-03

aiogoogle.validate.validate(instance, schema, schemas=None, schema_name=None)[source]

Parameters:

Instance – Instance to validate
schema – schema to validate instance against
schemas – Full schamas dict to resolve refs if any
schema_name – Name of the schema (Useful if you want more meaningful errors)

Sessions

class aiogoogle.sessions.aiohttp_session.AiohttpSession(*args, **kwargs)[source]: Bases: ClientSession, AbstractSession

class aiogoogle.sessions.curio_asks_session.CurioAsksSession(*args, **kwargs)[source]: Bases: Session, AbstractSession

class aiogoogle.sessions.trio_asks_session.TrioAsksSession(*args, **kwargs)[source]: Bases: Session, AbstractSession

Tips and hints

For a more efficient use of your quota use partial responses.

Contribute 🙋

There’s a bunch you can do to help regardless of your experience level:

Features, chores and bug reports:

Please refer to the Github issue tracker where they are posted.
Examples:

You can add examples to the examples folder
Testing:

Add more tests, the library is currently a bit undertested

Take your pick :)

Development notes

To run the test suite

Install tox
Run $ tox in the root of the project.
For extra configs, check out the tox.ini file.

To upload a new version

Bump the version

In: aiogoogle/__version__.py

Push a new tag

Make a new Git tag then push it.

$ git tag -a 1.0.1 -m "New version :tada:" master
$ git push --tags

Publish release

Then go to Github -> Releases -> Click on the tag -> Write a release title and description -> Hit the publish release button.

The publish.yml Github action should take it from there. It will build and then upload the new package to Pypi.

To install the local version of the aiogoogle instead of the latest one on Pip

$ pip uninstall aiogoogle
$ cd {cloned_aiogoogle_repo_with_your_local_changes}
$ pip install -e .

Now you can import aiogoogle from anywhere in your FS and make it use your local version

Aiogoogle

Library Setup ⚙️

Google Account Setup

Authentication

User account

OAuth2

Quick auth script

Integrate OAuth2 in an existing web app

OpenID Connect

Quick auth script

Integrate OpenID Connect in an existing web app

Service account

OAuth2

Others

Anonymous principal (API keys)

Discovery Service

Discovery docs and the Aiogoogle object explained

Intro

Structure of an API

Browse an API - Programatically

Browse an API - Manually

Send a Request

Send Requests Concurrently

Send As API key

Send As User

Examples

List Files on Google Drive

Pagination

Download a File from Google Drive

Upload a File to Google Drive

Upload a File to Google Drive using an AsyncIterable

List Your Contacts

List Calendar Events

Verify an Android In App Purchase

Design and goals

Async framework agnostic

Lightweight and minimalistic

API

Aiogoogle

GoogleAPI

Resource

Method

Credentials

Auth Managers

Exceptions

Models

Utils

Validate

Sessions

Tips and hints

Contribute 🙋

Development notes

To run the test suite

To upload a new version

To install the local version of the aiogoogle instead of the latest one on Pip

Discovery docs and the `Aiogoogle` object explained