gcloud-aio-auth

import asyncio import os import aiohttp from gcloud.aio.auth import Token async def main(): # Attempt to use GOOGLE_APPLICATION_CREDENTIALS environment variable or ADC # For local development, ensure `gcloud auth application-default login` has been run # or GOOGLE_APPLICATION_CREDENTIALS points to a service account key file. service_account_path = os.environ.get('GOOGLE_APPLICATION_CREDENTIALS', '') print("Initializing Token...") # Initialize Token; it will try to discover credentials if service_file is None. # Specify necessary scopes for your application. token = Token( service_file=service_account_path if service_account_path else None, scopes=["https://www.googleapis.com/auth/cloud-platform"] ) async with aiohttp.ClientSession() as session: try: # Get an access token (automatically refreshed by the Token instance) access_token = await token.get() print(f"Successfully obtained access token (first 10 chars): {access_token[:10]}...") # Example: Make an authenticated request to a Google Cloud API # (e.g., list buckets in Google Cloud Storage) headers = {"Authorization": f"Bearer {access_token}"} print("Making a dummy authenticated request to Google Cloud Storage API...") async with session.get( "https://www.googleapis.com/storage/v1/projects/_/buckets", headers=headers ) as response: if response.status == 200: print(f"Request to GCS successful (status 200). Some buckets (if any): {await response.json()}") else: print(f"Request failed with status: {response.status}") print(f"Response body: {await response.text()}") except Exception as e: print(f"An error occurred: {e}") finally: # Ensure the token's internal session is closed await token.close() if __name__ == "__main__": asyncio.run(main())