aiogoogle 0.1.17

Async Discovery API Client + Authentication

Aiogoogle

Async Discovery Service Client +

Async Google OAuth2 Client +

Async Google OpenID Connect (Social Sign-in) 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
• Kubernetes Engine API
• And more

Setup ⚙️

$ pip install aiogoogle

Google Account Setup

1. Create a project: Google’s APIs and Services dashboard
2. Enable an API: API Library
3. Create credentials: Credentials wizard
4. Pick an API: Google's API explorer

Authentication

There are 3 main authentication schemes you can use with Google's discovery service:

1. OAuth2

   Should be used whenever you want to access personal information from user accounts.

   Also, Aiogoogle supports Google OpenID connect which is a superset of OAuth2. (Google Social Signin)

2. API key

   Suitable when accessing Public information.

   A simple secret string, that you can get from Google's Cloud Console

   Note:

    For most personal information, an API key won't be enough.
   
    You should use OAuth2 instead.
   

3. Service Accounts

   A service account is a special kind of account that belongs to an application or a virtual machine (VM) instance, not a person.

   Note:

    Not yet supported by Aiogoogle
   

OAuth2 Primer

Oauth2 serves as an authorization framework. It supports 4 main flows:

1. Authorization code flow *Only flow suppoerted:

   • RFC6749 section 4.1

2. Client Credentials Flow:

   • Similar to the API key authentication scheme provided by Aiogoogle, so use it instead.
   • RFC6749 section 4.4

3. Implicit Grant Flow:

   • Not supported.
   • RFC6749 section 4.2

4. Resource Owner Password Credentials Flow:

   • Not supported.
   • RFC6749 section 4.3

Since Aiogoogle only supports Authorization Code Flow which happens to fit most use cases, let's dig a little in to it:

Authorization Code Flow

There are 3 main parties involved in this flow:

1. User:
   • The application consumer.
   • represented as aiogoogle.UserCreds
2. Client:
   • The Application owner.
   • represented as aiogoogle.ClientCreds
3. Resource Server:
   • The service that aiogoogle acts as a client to. e.g. Google Analytics, Youtube, etc.

Here's a nice ASCII chart showing how this flow works RFC6749 section 4.1 Figure 3

+----------+
| Resource |
|   Owner  |
|          |
+----------+
    ^
    |
    (B)
+----|-----+          Client Identifier      +---------------+
|         -+----(A)-- & Redirection URI ---->|               |
|  User-   |                                 | Authorization |
|  Agent  -+----(B)-- User authenticates --->|     Server    |
|          |                                 |               |
|         -+----(C)-- Authorization Code ---<|               |
+-|----|---+                                 +---------------+
|    |                                           ^      v
(A)  (C)                                         |      |
|    |                                           |      |
^    v                                           |      |
+---------+                                      |      |
|         |>---(D)-- Authorization Code ---------'      |
|  Client |          & Redirection URI                  |
|         |                                             |
|         |<---(E)----- Access Token -------------------'
+---------+       (w/ Optional Refresh Token)

Authorization examples (the examples require Sanic HTTP Server)

Get user credentials using OAuth2 (Authorization code flow) full example

import webbrowser

from sanic import Sanic, response
from sanic.exceptions import ServerError

from aiogoogle import Aiogoogle
from aiogoogle.auth.utils import create_secret


EMAIL = "client email"
CLIENT_CREDS = {
    "client_id": '...',
    "client_secret": '...',
    "scopes": ['...'],
    "redirect_uri": "http://localhost:5000/callback/aiogoogle",
}
state = create_secret()  # Shouldn't be a global hardcoded variable.

LOCAL_ADDRESS = "localhost"
LOCAL_PORT = "5000"

app = Sanic(__name__)
aiogoogle = Aiogoogle(client_creds=CLIENT_CREDS)


@app.route("/authorize")
def authorize(request):
    if aiogoogle.oauth2.is_ready(CLIENT_CREDS):
        uri = aiogoogle.oauth2.authorization_url(
            client_creds=CLIENT_CREDS,
            state=state,
            access_type="offline",
            include_granted_scopes=True,
            login_hint=EMAIL,
            prompt="select_account",
        )
        return response.redirect(uri)
    else:
        raise ServerError("Client doesn't have enough info for Oauth2")


@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"):
        returned_state = request.args["state"][0]
        if returned_state != state:
            raise ServerError("NO")
        full_user_creds = await aiogoogle.oauth2.build_user_creds(
            grant=request.args.get("code"), client_creds=CLIENT_CREDS
        )
        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")


if __name__ == "__main__":
    webbrowser.open("http://" + LOCAL_ADDRESS + ":" + LOCAL_PORT + "/authorize")
    app.run(host=LOCAL_ADDRESS, port=LOCAL_PORT, debug=True)

OpenID Connect (Social signin) full example

import webbrowser
import pprint

from sanic import Sanic, response
from sanic.exceptions import ServerError

from aiogoogle import Aiogoogle
from aiogoogle.auth.utils import create_secret

EMAIL = "..."
CLIENT_CREDS = {
    "client_id": "...",
    "client_secret": "...",
    "scopes": ["openid", "email"],
    "redirect_uri": "http://localhost:5000/callback/aiogoogle",
}
state = (
    create_secret()
)  # Shouldn't be a global or a hardcoded variable. should be tied to a session or a user and shouldn't be used more than once
nonce = (
    create_secret()
)  # Shouldn't be a global or a hardcoded variable. should be tied to a session or a user and shouldn't be used more than once


LOCAL_ADDRESS = "localhost"
LOCAL_PORT = "5000"

app = Sanic(__name__)
aiogoogle = Aiogoogle(client_creds=CLIENT_CREDS)


@app.route("/authorize")
def authorize(request):
    if aiogoogle.openid_connect.is_ready(CLIENT_CREDS):
        uri = aiogoogle.openid_connect.authorization_url(
            client_creds=CLIENT_CREDS,
            state=state,
            nonce=nonce,
            access_type="offline",
            include_granted_scopes=True,
            login_hint=EMAIL,
            prompt="select_account",
        )
        return response.redirect(uri)
    else:
        raise ServerError("Client doesn't have enough info for Oauth2")


@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"):
        returned_state = request.args["state"][0]
        if returned_state != state:
            raise ServerError("NO")
        full_user_creds = await aiogoogle.openid_connect.build_user_creds(
            grant=request.args.get("code"),
            client_creds=CLIENT_CREDS,
            nonce=nonce,
            verify=False,
        )
        full_user_info = await aiogoogle.openid_connect.get_user_info(full_user_creds)
        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")


if __name__ == "__main__":
    webbrowser.open("http://" + LOCAL_ADDRESS + ":" + LOCAL_PORT + "/authorize")
    app.run(host=LOCAL_ADDRESS, port=LOCAL_PORT, debug=True)

API key example

No need for an example because it's very simple. Just get an API key from your Google management console and pass it on to your Aiogoogle instance. Like this:

aiogoogle = Aiogoogle(api_key='...')

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

To understand how to navigate a discovery service/document and access the API endpoints that you desire using the Aiogoogle object, it is highly recommended that you read this section in the docs.

Quick Examples

List your Google Drive Files full example

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')
        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())

Shorten a URL using an API key as the authentication scheme of choice

import asyncio
from aiogoogle import Aiogoogle
from pprint import pprint

async def shorten_url(long_urls):
    async with Aiogoogle(api_key=api_key) as google:
        url_shortener = await google.discover('urlshortener', 'v1')
        short_urls = await google.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/"
    }
]

List your Google Calendar events using Trio | full example

$ pip install aiogoogle[trio_asks]

import trio
from aiogoogle import Aiogoogle
from aiogoogle.sessions.trio_asks_session import TrioAsksSession


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

async def list_events():
    async with Aiogoogle(
        user_creds=user_creds,
        session_factory=TrioAsksSession,
    ) as aiogoogle:
        calendar_v3 = await aiogoogle.discover("calendar", "v3")
        events = await aiogoogle.as_user(
            calendar_v3.events.list(calendarId="primary"), full_res=True
        )
    async for page in events:
        print(page)

trio.run(list_events)

List your Youtube videos using Curio | full example

$ pip install aiogoogle[curio_asks]

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


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

async def list_playlists():
    async with Aiogoogle(
        user_creds=user_creds,
        session_factory=CurioAsksSession,
    ) as aiogoogle:
        youtube_v3 = await aiogoogle.discover("youtube", "v3")
        req = youtube_v3.playlists.list(part="snippet", mine=True)
        res = await aiogoogle.as_user(req)
    print(res)

curio.run(list_playlists())

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())

Documentation 📑

readthedocs: https://aiogoogle.readthedocs.io/en/latest/

Contribute 🙋

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

1. Features, chores and bug reports:

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

2. Examples:

   You can add examples to the examples folder

3. Testing:

   Add more tests. The library is currently a bit undertested