aiogoogle

Use OpenID Connect (OIDC) if you need to identify the user (e.g., for 'Social Sign-in'). OIDC is a superset of OAuth2; it provides the same access tokens but also returns an ID token containing user 'claims' (like sub, email, etc.).

1. Generate Authorization URL

Use aiogoogle.openid_connect.authorization_url. You should provide a nonce to prevent replay attacks.

2. Handle the Callback

Exchange the grant code for UserCreds and then use aiogoogle.openid_connect.get_user_info to retrieve the user's identity claims.

Note: Use the sub claim to uniquely identify users. Do not rely on email for unique identification, as emails can change.

# 1. Generate URL with nonce
uri = aiogoogle.openid_connect.authorization_url(
    client_creds={
        'client_id': '...',
        'client_secret': '...',
        'redirect_uri': '...',
        'scopes': ['...']
    },
    nonce='random_nonce_value'
)

# 2. Handle callback
full_user_creds = await aiogoogle.openid_connect.build_user_creds(
    grant=request.args.get('code'),
    client_creds=CLIENT_CREDS,
    nonce='random_nonce_value',
    verify=False
)

# 3. Get user identity claims
full_user_info = await aiogoogle.openid_connect.get_user_info(full_user_creds)
# full_user_info contains claims like 'sub', 'email', etc.

You can point the GOOGLE_APPLICATION_CREDENTIALS environment variable to your JSON key file location. To make Aiogoogle load these credentials, you must call await aiogoogle.service_account_manager.detect_default_creds_source() before making API calls.

export GOOGLE_APPLICATION_CREDENTIALS="location_of_key_file.json"

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

creds = ServiceAccountCreds(scopes=["https://www.googleapis.com/auth/cloud-platform"])

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

    # Load the service account key from the environment variable
    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())

You can authenticate using a user-managed service account JSON key file by passing its contents to ServiceAccountCreds. This is useful for applications that need to perform actions on their own. When using this method, use aiogoogle.as_service_account() to execute the discovered API calls.

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

# Load the JSON key file
service_account_key = json.load(open('test_service_account.json'))

# Initialize credentials with required scopes
creds = ServiceAccountCreds(
    scopes=[
        "https://www.googleapis.com/auth/devstorage.read_only",
        "https://www.googleapis.com/auth/devstorage.read_write",
    ],
    **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")
        # Use as_service_account to wrap the API call
        res = await aiogoogle.as_service_account(
            storage.buckets.list(project=creds["project_id"])
        )
        print(res)

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

Once you have discovered an API using await aiogoogle.discover(name, version), you can inspect its structure programmatically:

1. List Resources: Access api_object['resources'].
2. Access a Resource: Use dot notation (e.g., api_object.resource_name).
3. Check Available Methods: Use resource.methods_available.
4. Inspect a Method: Access the method object (e.g., resource.method_name) to see:
   • description: A string describing the method.
   • optional_parameters: A list of parameter names.
   • required_parameters: A list of parameter names.
   • parameters: A dictionary containing detailed metadata (type, description, location) for each parameter.

# Example: Inspecting a method
url_resource = url_shortener.url
print(url_resource.methods_available)  # ['get', 'insert', 'list']

list_url = url_resource.list
print(list_url.optional_parameters)    # ['projection', 'start_token', ...]
print(list_url.parameters['start_token'])